"""SQL generation using SQLGlot builder API."""

import logging

import sqlglot
from sqlglot import exp, select

from sidemantic.core.preagg_matcher import PreAggregationMatcher
from sidemantic.core.semantic_graph import SemanticGraph
from sidemantic.core.symmetric_aggregate import build_symmetric_aggregate_sql
from sidemantic.sql.aggregation_detection import sql_has_aggregate


class SQLGenerator:
    """Generates SQL queries from semantic layer definitions using SQLGlot builder API."""

    def __init__(
        self,
        graph: SemanticGraph,
        dialect: str = "duckdb",
        preagg_database: str | None = None,
        preagg_schema: str | None = None,
    ):
        """Initialize SQL generator.

        Args:
            graph: Semantic graph with models and metrics
            dialect: SQL dialect for generation (default: duckdb)
            preagg_database: Optional database name for pre-aggregation tables
            preagg_schema: Optional schema name for pre-aggregation tables
        """
        self.graph = graph
        self.dialect = dialect
        self.preagg_database = preagg_database
        self.preagg_schema = preagg_schema
        self._generate_cache: dict[tuple[object, ...], str] = {}
        self._generate_cache_limit = 256

    @staticmethod
    def _agg_sql_name(agg: str) -> str:
        """Return the SQL function name for a normalized metric aggregation."""
        return {"variance_pop": "VAR_POP", "var_pop": "VAR_POP"}.get(agg, agg.upper())

    @staticmethod
    def _model_from_clause(model) -> str:
        if model.sql:
            return f"({model.sql}) AS t"
        if model.table:
            return model.table
        if model.source_uri:
            raise ValueError(
                f"Model '{model.name}' uses source_uri '{model.source_uri}', but Python SQL generation does not "
                "load source_uri data. Define table or sql for query compilation."
            )
        raise ValueError(f"Model '{model.name}' must define table or sql for query compilation")

    def _model_source_as(self, model, alias: str) -> str:
        quoted_alias = self._quote_identifier(alias)
        if model.sql:
            return f"({model.sql}) AS {quoted_alias}"
        if model.table:
            return f"{model.table} AS {quoted_alias}"
        if model.source_uri:
            raise ValueError(
                f"Model '{model.name}' uses source_uri '{model.source_uri}', but Python SQL generation does not "
                "load source_uri data. Define table or sql for query compilation."
            )
        raise ValueError(f"Model '{model.name}' must define table or sql for query compilation")

    @staticmethod
    def _freeze_cache_value(value):
        if isinstance(value, dict):
            items = (
                (SQLGenerator._freeze_cache_value(k), SQLGenerator._freeze_cache_value(v)) for k, v in value.items()
            )
            return tuple(sorted(items, key=repr))
        if isinstance(value, (list, tuple)):
            return tuple(SQLGenerator._freeze_cache_value(item) for item in value)
        if isinstance(value, set):
            return tuple(sorted((SQLGenerator._freeze_cache_value(item) for item in value), key=repr))
        try:
            hash(value)
            return value
        except TypeError:
            return repr(value)

    def _generate_cache_key(
        self,
        metrics,
        dimensions,
        filters,
        segments,
        order_by,
        limit,
        offset,
        parameters,
        ungrouped,
        use_preaggregations,
        aliases,
        skip_default_time_dimensions,
    ) -> tuple[object, ...]:
        return (
            getattr(self.graph, "_version", 0),
            self.dialect,
            self.preagg_database,
            self.preagg_schema,
            self._freeze_cache_value(metrics),
            self._freeze_cache_value(dimensions),
            self._freeze_cache_value(filters),
            self._freeze_cache_value(segments),
            self._freeze_cache_value(order_by),
            limit,
            offset,
            self._freeze_cache_value(parameters),
            ungrouped,
            use_preaggregations,
            self._freeze_cache_value(aliases),
            skip_default_time_dimensions,
        )

    def _cache_generate_result(self, cache_key: tuple[object, ...], sql: str) -> str:
        if len(self._generate_cache) >= self._generate_cache_limit:
            self._generate_cache.pop(next(iter(self._generate_cache)))
        self._generate_cache[cache_key] = sql
        return sql

    def _date_trunc(self, granularity: str, column_expr: str) -> str:
        """Generate dialect-specific time truncation expression.

        Args:
            granularity: Time granularity (hour, day, week, month, quarter, year)
            column_expr: SQL column expression

        Returns:
            Time truncation SQL expression appropriate for the dialect
        """
        if self.dialect == "bigquery":
            return f"DATE_TRUNC({column_expr}, {granularity.upper()})"

        if self.dialect in {"spark", "databricks"}:
            spark_granularity = granularity.upper()
            if spark_granularity in {"WEEK", "MONTH", "QUARTER", "YEAR"}:
                return f"TRUNC({column_expr}, '{spark_granularity}')"
            if spark_granularity == "DAY":
                return f"CAST(DATE_TRUNC('DAY', {column_expr}) AS DATE)"
            return f"DATE_TRUNC('{spark_granularity}', {column_expr})"

        # Handle {model} placeholder or complex expressions - fall back to string
        if "{" in column_expr or "(" in column_expr:
            return f"DATE_TRUNC('{granularity}', {column_expr})"

        # Parse the column expression to handle table.column references
        col = sqlglot.parse_one(column_expr, into=exp.Column, dialect=self.dialect)
        date_trunc = exp.DateTrunc(this=col, unit=exp.Literal.string(granularity))
        return date_trunc.sql(dialect=self.dialect)

    def _build_interval(self, num: str, unit: str) -> str:
        """Build dialect-specific INTERVAL expression.

        Args:
            num: Numeric value (e.g., "7")
            unit: Time unit (e.g., "days")

        Returns:
            SQL INTERVAL expression appropriate for the dialect
        """
        if self.dialect == "bigquery":
            # BigQuery: INTERVAL 7 DAY (no quotes, singular unit)
            singular_unit = unit.upper().rstrip("S")
            return f"INTERVAL {num} {singular_unit}"
        else:
            # Standard SQL: INTERVAL '7 days'
            return f"INTERVAL '{num} {unit}'"

    def _date_diff(self, later: str, earlier: str, unit: str = "day") -> str:
        """Generate dialect-specific date difference expression.

        Args:
            later: SQL expression for the later date
            earlier: SQL expression for the earlier date
            unit: Time unit for the difference (e.g., "day")

        Returns:
            SQL expression computing the difference in the given unit
        """
        if self.dialect == "bigquery":
            return f"DATE_DIFF({later}, {earlier}, {unit.upper()})"
        elif self.dialect == "snowflake":
            return f"DATEDIFF('{unit}', {earlier}, {later})"
        else:
            # DuckDB and Postgres support direct date subtraction
            return f"({later} - {earlier})"

    def _strip_model_prefixes(self, filters: list[str], model_name: str) -> list[str]:
        """Strip model name prefixes from filter column references.

        Retention and conversion CTEs query the raw table directly, so qualified
        references like ``events.region`` need to become just ``region``.

        Args:
            filters: List of filter SQL expressions
            model_name: Model name whose prefix should be stripped

        Returns:
            Filters with model prefixes removed
        """
        result = []
        for f in filters:
            # Resolve {model} placeholders to the actual model name before
            # parsing, so sqlglot can handle the expression as valid SQL.
            f_resolved = f.replace("{model}.", f"{model_name}.")
            try:
                parsed = sqlglot.parse_one(f_resolved, dialect=self.dialect)
                for column in parsed.find_all(exp.Column):
                    tbl = column.table
                    if tbl and tbl.replace("_cte", "") == model_name:
                        column.set("table", None)
                result.append(parsed.sql(dialect=self.dialect))
            except Exception:
                result.append(f_resolved)
        return result

    def _resolve_filter_dimensions(self, filters: list[str], model) -> list[str]:
        """Resolve semantic dimension names in filter expressions to their SQL expressions.

        After ``_strip_model_prefixes`` turns ``events.city`` into ``city``,
        this method rewrites ``city`` to the actual SQL expression
        (e.g., ``json_extract_string(properties, '$."$geoip_city_name"')``).

        This is needed because retention, conversion, and cohort CTEs query
        the raw table directly, where semantic dimension names don't exist
        as columns.

        Args:
            filters: Filter expressions with model prefixes already stripped
            model: The model whose dimensions to resolve

        Returns:
            Filters with dimension names replaced by their SQL expressions
        """
        dim_map = {}
        for dim in model.dimensions:
            if dim.sql_expr != dim.name:
                dim_map[dim.name] = dim.sql_expr

        if not dim_map:
            return filters

        result = []
        for f in filters:
            try:
                parsed = sqlglot.parse_one(f, dialect=self.dialect)
                for column in parsed.find_all(exp.Column):
                    if not column.table and column.name in dim_map:
                        replacement = sqlglot.parse_one(dim_map[column.name], dialect=self.dialect)
                        # Strip model-qualified refs from the replacement so
                        # expressions like events.region don't leak into
                        # subquery contexts where the alias is t/s.
                        for rcol in replacement.find_all(exp.Column):
                            if rcol.table and rcol.table.replace("_cte", "") == model.name:
                                rcol.set("table", None)
                        column.replace(replacement)
                result.append(parsed.sql(dialect=self.dialect))
            except Exception:
                result.append(f)
        return result

    def _quote_alias(self, name: str) -> str:
        """Quote an identifier for use as a SQL alias.

        Handles names with dots or other special characters by quoting them.

        Args:
            name: The identifier name (e.g., "auctions.bid_request_cnt_wow")

        Returns:
            Properly quoted identifier for the dialect (e.g., '"auctions.bid_request_cnt_wow"')
        """
        return self._quote_identifier(name)

    @staticmethod
    def _is_simple_identifier(name: str) -> bool:
        """Return True when identifier can be emitted without quotes."""
        import re

        return re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name) is not None

    def _quote_identifier(self, name: str) -> str:
        """Quote a SQL identifier for the current dialect.

        Delegates to sqlglot which handles reserved words (e.g., 'order')
        and special characters automatically.
        """
        if self._is_simple_identifier(name):
            return sqlglot.to_identifier(name, quoted=False).sql(dialect=self.dialect)
        return sqlglot.to_identifier(name, quoted=True).sql(dialect=self.dialect)

    def _cte_name(self, model_name: str) -> str:
        """Get the CTE identifier name for a model."""
        return f"{model_name}_cte"

    def _cte_ref(self, model_name: str, column_name: str) -> str:
        """Build a quoted reference to a CTE column."""
        return f"{self._quote_identifier(self._cte_name(model_name))}.{self._quote_identifier(column_name)}"

    def _custom_join_condition(self, join_path) -> str:
        """Render a relationship custom SQL join condition for generated CTE names."""
        from_alias = self._quote_identifier(self._cte_name(join_path.from_model))
        to_alias = self._quote_identifier(self._cte_name(join_path.to_model))
        return join_path.custom_condition.replace("{from}", from_alias).replace("{to}", to_alias)

    def _custom_join_columns(self, join_path) -> dict[str, set[str]]:
        """Extract raw columns that a custom join predicate reads from each side."""
        if not join_path.custom_condition:
            return {}

        from_marker = "__from__"
        to_marker = "__to__"
        condition = join_path.custom_condition.replace("{from}", from_marker).replace("{to}", to_marker)
        try:
            parsed = sqlglot.parse_one(condition, dialect=self.dialect)
        except Exception as exc:
            raise ValueError(
                "Could not parse custom relationship SQL for "
                f"{join_path.from_model} -> {join_path.to_model}: {join_path.custom_condition}"
            ) from exc

        columns: dict[str, set[str]] = {join_path.from_model: set(), join_path.to_model: set()}
        for column in parsed.find_all(exp.Column):
            if column.table == from_marker:
                columns[join_path.from_model].add(column.name)
            elif column.table == to_marker:
                columns[join_path.to_model].add(column.name)

        return {model_name: cols for model_name, cols in columns.items() if cols}

    def _custom_join_columns_by_model(self, base_model_name: str, other_models: list[str]) -> dict[str, set[str]]:
        columns_by_model: dict[str, set[str]] = {}
        for other_model in other_models:
            join_path = self.graph.find_relationship_path(base_model_name, other_model)
            if not join_path:
                continue
            for join_step in join_path:
                for model_name, columns in self._custom_join_columns(join_step).items():
                    columns_by_model.setdefault(model_name, set()).update(columns)
        return columns_by_model

    def _apply_default_time_dimensions(self, metrics: list[str], dimensions: list[str]) -> list[str]:
        """Auto-include default_time_dimension from models if not already present.

        If a model has default_time_dimension set and no time dimension from that
        model is already in the dimensions list, add it with the default_grain.

        Args:
            metrics: List of metric references
            dimensions: List of dimension references

        Returns:
            Updated dimensions list with default time dimensions added
        """
        # Extract which models already have time dimensions in the query
        models_with_time_dims = set()
        for dim_ref in dimensions:
            if "." in dim_ref:
                model_name, dim_part = dim_ref.split(".", 1)
                # Strip granularity suffix if present
                dim_name = dim_part.split("__")[0]
                model = self.graph.get_model(model_name)
                if model:
                    dim = model.get_dimension(dim_name)
                    if dim and dim.type == "time":
                        models_with_time_dims.add(model_name)

        # Check each model referenced by metrics for default_time_dimension
        added_dims = []
        models_checked = set()
        for metric_ref in metrics:
            try:
                model_name, _ = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                continue
            if not model_name or model_name in models_checked:
                continue
            models_checked.add(model_name)

            try:
                model = self.graph.get_model(model_name)
            except KeyError:
                model = None
            if model and model.default_time_dimension:
                # Only add if this model doesn't already have a time dimension
                if model_name not in models_with_time_dims:
                    time_dim_ref = f"{model_name}.{model.default_time_dimension}"
                    # Apply default_grain if specified
                    if model.default_grain:
                        time_dim_ref = f"{time_dim_ref}__{model.default_grain}"
                    if time_dim_ref not in dimensions and time_dim_ref not in added_dims:
                        added_dims.append(time_dim_ref)
                    models_with_time_dims.add(model_name)

        return dimensions + added_dims

    def generate_view(
        self,
        view_name: str,
        metrics: list[str] | None = None,
        dimensions: list[str] | None = None,
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
    ) -> str:
        """Generate CREATE VIEW statement for a semantic query.

        Allows joining arbitrary SQL against semantic layer calculations.

        Args:
            view_name: Name for the view
            metrics: List of metric references
            dimensions: List of dimension references
            filters: List of filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows

        Returns:
            CREATE VIEW SQL statement
        """
        import re

        if not re.match(r"^[a-zA-Z_][a-zA-Z0-9_.]*$", view_name):
            raise ValueError(f"Invalid view name: {view_name}")
        query_sql = self.generate(metrics, dimensions, filters, order_by, limit)
        return f"CREATE VIEW {view_name} AS\n{query_sql}"

    def generate(
        self,
        metrics: list[str] | None = None,
        dimensions: list[str] | None = None,
        filters: list[str] | None = None,
        segments: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
        parameters: dict[str, any] | None = None,
        ungrouped: bool = False,
        use_preaggregations: bool = False,
        aliases: dict[str, str] | None = None,
        skip_default_time_dimensions: bool = False,
    ) -> str:
        """Generate SQL query from semantic layer query.

        Args:
            metrics: List of metric references (e.g., ["orders.revenue"])
            dimensions: List of dimension references (e.g., ["orders.status", "orders.order_date__month"])
            filters: List of filter expressions (e.g., ["orders.status = 'completed'"])
            segments: List of segment references (e.g., ["orders.active_users"])
            order_by: List of fields to order by
            limit: Maximum number of rows to return
            offset: Number of rows to skip
            parameters: User-provided parameter values for interpolation
            ungrouped: If True, return raw rows without aggregation (no GROUP BY)
            use_preaggregations: Enable automatic pre-aggregation routing (default: False)
            aliases: Custom aliases for fields (dict mapping field reference to alias)
            skip_default_time_dimensions: If True, don't auto-include default_time_dimension

        Returns:
            SQL query string
        """
        metrics = metrics or []
        dimensions = list(dimensions) if dimensions else []
        filters = filters or []
        segments = segments or []
        parameters = parameters or {}
        aliases = aliases or {}
        cache_key = self._generate_cache_key(
            metrics,
            dimensions,
            filters,
            segments,
            order_by,
            limit,
            offset,
            parameters,
            ungrouped,
            use_preaggregations,
            aliases,
            skip_default_time_dimensions,
        )
        cached = self._generate_cache.get(cache_key)
        if cached is not None:
            return cached

        # Auto-include default_time_dimension from metrics if not already present
        if not skip_default_time_dimensions:
            dimensions = self._apply_default_time_dimensions(metrics, dimensions)

        # Resolve segments to SQL filters
        segment_filters = self._resolve_segments(segments)
        filters = filters + segment_filters

        # Interpolate parameters into filters if provided
        from sidemantic.core.parameter import ParameterSet

        param_set = ParameterSet(self.graph.parameters, parameters)
        filters = [param_set.interpolate(f) for f in filters]

        # Process relative date expressions in filters
        from sidemantic.core.relative_date import RelativeDateRange

        processed_filters = []
        for f in filters:
            # Check if filter contains a relative date expression
            # Pattern: column_name operator 'relative_date_expr'
            # e.g., "created_at >= 'last 7 days'"
            import re

            match = re.match(r"^(.+?)\s*(>=|<=|>|<|=)\s*['\"](.+?)['\"]$", f)
            if match:
                column, operator, value = match.groups()
                if RelativeDateRange.is_relative_date(value):
                    # Convert relative date to SQL expression
                    if operator in [">=", ">"]:
                        # For >= or >, just use the start date
                        sql_expr = RelativeDateRange.parse(value, self.dialect)
                        processed_filters.append(f"{column} {operator} {sql_expr}")
                    elif operator == "=":
                        # For =, use to_range to get proper range
                        range_expr = RelativeDateRange.to_range(value, column.strip(), self.dialect)
                        if range_expr:
                            processed_filters.append(range_expr)
                        else:
                            processed_filters.append(f)
                    else:
                        processed_filters.append(f)
                else:
                    processed_filters.append(f)
            else:
                processed_filters.append(f)

        filters = processed_filters

        # Check if any metrics need window functions (cumulative or time_comparison)
        def metric_needs_window(m):
            # Try to get metric - could be model.measure or just metric name
            metric = None
            try:
                _, metric = self.graph.resolve_metric_reference(m)
            except KeyError:
                pass

            if not metric and "." not in m:
                # Just metric name - try graph-level metric
                try:
                    metric = self.graph.get_metric(m)
                except KeyError:
                    pass
                # Fall back to scanning models (e.g. model-scoped metrics)
                if not metric:
                    for model in self.graph.models.values():
                        found = model.get_metric(m)
                        if found:
                            metric = found
                            break

            if not metric:
                return False

            # Cumulative and time_comparison always need windows
            if metric.type in ("cumulative", "time_comparison"):
                return True
            # Ratio with offset_window needs window
            if metric.type == "ratio" and metric.offset_window:
                return True
            # Conversion metrics need special handling
            if metric.type == "conversion":
                return True
            # Retention metrics need special handling
            if metric.type == "retention":
                return True
            # Cohort metrics need special handling
            if metric.type == "cohort":
                return True
            return False

        needs_window_functions = any(metric_needs_window(m) for m in metrics)

        if needs_window_functions:
            return self._cache_generate_result(
                cache_key,
                self._generate_with_window_functions(
                    metrics,
                    dimensions,
                    filters,
                    order_by,
                    limit,
                    offset,
                    aliases,
                    use_preaggregations=use_preaggregations,
                ),
            )

        # Parse dimension references and extract granularities
        parsed_dims = self._parse_dimension_refs(dimensions)

        # Find all models needed for the query
        model_names = self._find_required_models(metrics, dimensions, filters)

        if use_preaggregations and not ungrouped:
            join_key_preagg_sql = self._try_use_join_key_preaggregation(
                metrics=metrics,
                dimensions=dimensions,
                filters=filters,
                order_by=order_by,
                limit=limit,
                offset=offset,
                aliases=aliases,
            )
            if join_key_preagg_sql:
                instrumentation = self._generate_instrumentation_comment(
                    models=model_names, metrics=metrics, dimensions=dimensions, used_preagg=True
                )
                return self._cache_generate_result(cache_key, join_key_preagg_sql + "\n" + instrumentation)

        # Check if we need symmetric aggregation (pre-aggregation approach)
        # This is needed when metrics come from different models at different join levels
        if self._needs_preaggregation_for_fanout(metrics, dimensions):
            return self._cache_generate_result(
                cache_key,
                self._generate_with_preaggregation(
                    metrics=metrics,
                    dimensions=dimensions,
                    filters=filters,
                    segments=None,  # Already resolved into filters above
                    order_by=order_by,
                    limit=limit,
                    offset=offset,
                    aliases=aliases,
                    use_preaggregations=use_preaggregations,
                ),
            )

        # Try to use pre-aggregation if enabled (single model queries only)
        if use_preaggregations and len(model_names) == 1 and not ungrouped:
            preagg_sql = self._try_use_preaggregation(
                model_name=model_names[0],
                metrics=metrics,
                parsed_dims=parsed_dims,
                filters=filters,
                order_by=order_by,
                limit=limit,
                offset=offset,
                aliases=aliases,
            )
            if preagg_sql:
                # Add instrumentation comment
                instrumentation = self._generate_instrumentation_comment(
                    models=[model_names[0]], metrics=metrics, dimensions=dimensions, used_preagg=True
                )
                return self._cache_generate_result(cache_key, preagg_sql + "\n" + instrumentation)

        if not model_names:
            raise ValueError("No models found for query")

        # Determine base model (first model in query - order is now preserved)
        base_model_name = model_names[0]

        # Find all intermediate models needed for joins
        all_models = set(model_names)
        for i, model_a in enumerate(list(model_names)):
            for model_b in list(model_names)[i + 1 :]:
                # Find join path and add intermediate models
                try:
                    join_path = self.graph.find_relationship_path(model_a, model_b)
                    for jp in join_path:
                        all_models.add(jp.from_model)
                        all_models.add(jp.to_model)
                except ValueError:
                    # No join path found
                    pass

        # Classify filters for pushdown optimization
        pushdown_filters, main_query_filters, window_dim_filters = self._classify_filters_for_pushdown(
            filters or [], all_models
        )
        # In the standard (non-preagg) path, window-dim filters belong in the
        # outer query WHERE, so merge them into main_query_filters.
        for wf_list in window_dim_filters.values():
            main_query_filters.extend(wf_list)

        # Determine which models have filters (for join type decision)
        models_with_filters = set()
        for model_name, model_filters in pushdown_filters.items():
            if model_filters:
                models_with_filters.add(model_name)
        # Also check main query filters
        for filter_expr in main_query_filters:
            try:
                parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
                for column in parsed.find_all(exp.Column):
                    if column.table:
                        # Remove _cte suffix if present
                        model_name = column.table.replace("_cte", "")
                        models_with_filters.add(model_name)
            except Exception:
                logging.debug("Failed to parse filter for model extraction: %s", filter_expr, exc_info=True)

        # Extract columns needed for metric-level filters (before building CTEs)
        metric_filter_cols_by_model = self._extract_metric_filter_columns(metrics)
        custom_join_cols_by_model = self._custom_join_columns_by_model(base_model_name, model_names[1:])
        for model_name, column_names in custom_join_cols_by_model.items():
            metric_filter_cols_by_model.setdefault(model_name, set()).update(column_names)

        # Ensure dimensions referenced in outer-query filters (e.g. window dims)
        # are included in the relevant CTE SELECT lists.
        for filter_expr in main_query_filters:
            try:
                parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
                for column in parsed.find_all(exp.Column):
                    if column.table:
                        model_name = column.table.replace("_cte", "")
                        if model_name in all_models:
                            if model_name not in metric_filter_cols_by_model:
                                metric_filter_cols_by_model[model_name] = set()
                            metric_filter_cols_by_model[model_name].add(column.name)
            except Exception:
                logging.debug(
                    "Failed to parse outer filter for column extraction: %s",
                    filter_expr,
                    exc_info=True,
                )

        # Build CTEs for all models with pushed-down filters
        cte_sqls = []
        for model_name in all_models:
            model_filters = pushdown_filters.get(model_name, [])
            metric_filter_cols = metric_filter_cols_by_model.get(model_name)
            cte_sql = self._build_model_cte(
                model_name,
                parsed_dims,
                metrics,
                model_filters if model_filters else None,
                order_by=order_by,
                all_models=all_models,
                metric_filter_columns=metric_filter_cols,
                ungrouped=ungrouped,
            )
            cte_sqls.append(cte_sql)

        # Build main SELECT using builder API (only with filters that couldn't be pushed down)
        query = self._build_main_select(
            base_model_name=base_model_name,
            other_models=model_names[1:] if len(model_names) > 1 else [],
            parsed_dims=parsed_dims,
            metrics=metrics,
            filters=main_query_filters,
            models_with_filters=models_with_filters,
            order_by=order_by,
            limit=limit,
            offset=offset,
            ungrouped=ungrouped,
            aliases=aliases,
        )

        # Combine CTEs and main query
        if cte_sqls:
            # Build WITH clause manually as string since builder API doesn't support CTEs well
            cte_str = "WITH " + ",\n".join(cte_sqls)
            full_sql = cte_str + "\n" + query
        else:
            full_sql = query

        # Add instrumentation comment for query analysis
        instrumentation = self._generate_instrumentation_comment(
            models=list(all_models), metrics=metrics, dimensions=dimensions, used_preagg=False
        )
        full_sql = full_sql + "\n" + instrumentation

        return self._cache_generate_result(cache_key, full_sql)

    def _parse_dimension_refs(self, dimensions: list[str]) -> list[tuple[str, str | None]]:
        """Parse dimension references to extract granularities.

        Args:
            dimensions: List of dimension references (e.g., ["orders.order_date__month"])

        Returns:
            List of (dimension_ref, granularity) tuples
        """
        parsed = []
        for dim in dimensions:
            if "__" in dim:
                dim_ref, gran = dim.rsplit("__", 1)
                parsed.append((dim_ref, gran))
            else:
                parsed.append((dim, None))
        return parsed

    def _resolve_segments(self, segments: list[str]) -> list[str]:
        """Resolve segment references to SQL filter expressions.

        Args:
            segments: List of segment references (e.g., ["orders.active_users"])

        Returns:
            List of SQL filter expressions

        Raises:
            ValueError: If segment not found
        """

        def qualify_unaliased_columns(filter_sql: str, model_alias: str) -> str:
            """Qualify unaliased columns in segment filters with model alias."""
            try:
                parsed = sqlglot.parse_one(filter_sql, dialect=self.dialect)
            except Exception:
                return filter_sql

            def visit(node: exp.Expression) -> None:
                if isinstance(node, exp.Subquery):
                    return

                if isinstance(node, exp.Column) and not node.table:
                    node.set("table", model_alias)

                for arg in node.args.values():
                    if isinstance(arg, exp.Expression):
                        visit(arg)
                    elif isinstance(arg, list):
                        for item in arg:
                            if isinstance(item, exp.Expression):
                                visit(item)

            visit(parsed)

            return parsed.sql(dialect=self.dialect)

        filters = []
        for seg_ref in segments:
            # Parse model.segment format
            if "." not in seg_ref:
                raise ValueError(f"Segment reference must be in format 'model.segment': {seg_ref}")

            model_name, segment_name = seg_ref.split(".", 1)
            model = self.graph.get_model(model_name)
            if not model:
                raise ValueError(f"Model '{model_name}' not found for segment '{seg_ref}'")

            segment = model.get_segment(segment_name)
            if not segment:
                raise ValueError(f"Segment '{segment_name}' not found on model '{model_name}'")

            # Get SQL expression with model alias replaced
            # Use model_cte as the alias (consistent with CTE naming)
            filter_sql = segment.get_sql(f"{model_name}_cte")
            filter_sql = qualify_unaliased_columns(filter_sql, f"{model_name}_cte")
            filters.append(filter_sql)

        return filters

    def _extract_models_from_sql(self, sql_expr: str) -> set[str]:
        """Extract referenced model names from qualified column references."""
        models: set[str] = set()
        try:
            parsed = sqlglot.parse_one(sql_expr, dialect=self.dialect)
            for column in parsed.find_all(exp.Column):
                if not column.table:
                    continue
                model_name = column.table.replace("_cte", "")
                if model_name in self.graph.models:
                    models.add(model_name)
        except Exception:
            pass
        return models

    def _find_required_models(
        self, metrics: list[str], dimensions: list[str], filters: list[str] | None = None
    ) -> list[str]:
        """Find all models required for the query.

        Args:
            metrics: List of metric references
            dimensions: List of dimension references
            filters: Optional list of filter expressions

        Returns:
            List of model names in the order they are encountered (preserves order)
        """
        models = []
        seen = set()

        def add_model(model_name: str):
            """Add model if not seen yet."""
            if model_name not in seen:
                models.append(model_name)
                seen.add(model_name)

        def collect_models_from_metric(metric_ref: str):
            """Recursively collect models needed from a metric."""
            try:
                model_name, metric = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                return

            if model_name:
                add_model(model_name)
                if metric.type == "ratio":
                    if metric.numerator:
                        collect_models_from_metric(metric.numerator)
                    if metric.denominator:
                        collect_models_from_metric(metric.denominator)
                elif metric.type == "derived" or (not metric.type and not metric.agg and metric.sql):
                    for ref_metric in metric.get_dependencies(self.graph, model_name):
                        collect_models_from_metric(ref_metric)
                    if metric.sql:
                        for ref_model_name in self._extract_models_from_sql(metric.sql):
                            add_model(ref_model_name)
                elif metric.agg and metric.sql:
                    for ref_model_name in self._extract_models_from_sql(metric.sql):
                        add_model(ref_model_name)
                return

            # It's a graph-level metric, need to resolve its dependencies.
            if metric.type == "ratio":
                if metric.numerator:
                    collect_models_from_metric(metric.numerator)
                if metric.denominator:
                    collect_models_from_metric(metric.denominator)
            elif metric.type == "derived" or (not metric.type and not metric.agg and metric.sql):
                # Derived or untyped metrics with sql - auto-detect dependencies
                for ref_metric in metric.get_dependencies(self.graph):
                    collect_models_from_metric(ref_metric)
                # Inline SQL expression metrics (e.g., SUM(orders.amount))
                # can have empty dependencies, so also parse model refs directly.
                if metric.sql:
                    for model_name in self._extract_models_from_sql(metric.sql):
                        add_model(model_name)
            elif metric.agg and metric.sql:
                # Graph-level simple aggregations can qualify fields
                # (e.g., SUM(orders.amount)); include those models.
                for model_name in self._extract_models_from_sql(metric.sql):
                    add_model(model_name)

        # Collect from dimensions first (since they define the grain)
        for dim in dimensions:
            # Remove granularity suffix if present
            if "__" in dim:
                dim = dim.rsplit("__", 1)[0]
            if "." in dim:
                add_model(dim.split(".")[0])

        # Then collect from metrics
        for metric in metrics:
            collect_models_from_metric(metric)

        # Finally, collect from filters (to support filtering on joined tables without selecting dimensions)
        if filters:
            for filter_expr in filters:
                # Parse filter to find model references
                try:
                    parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
                    # Find all column references in the filter
                    for column in parsed.find_all(exp.Column):
                        if column.table:
                            # Remove _cte suffix if present (shouldn't be, but be defensive)
                            model_name = column.table.replace("_cte", "")
                            add_model(model_name)
                except Exception:
                    logging.debug("Failed to parse filter for model extraction: %s", filter_expr, exc_info=True)

        return models

    def _classify_filters_for_pushdown(
        self, filters: list[str], all_models: set[str]
    ) -> tuple[dict[str, list[str]], list[str], dict[str, list[str]]]:
        """Classify filters into those that can be pushed down vs those that must stay in main query.

        Compound AND expressions are split into individual conjuncts first so that
        ``orders.status = 'active' AND items.category = 'A'`` gets decomposed into
        two single-model filters that can each be pushed down, rather than being kept
        as an un-pushable cross-model filter.

        Args:
            filters: List of filter expressions
            all_models: Set of all model names in the query

        Returns:
            Tuple of (pushdown_filters_by_model, main_query_filters, window_dim_filters_by_model)
            - pushdown_filters_by_model: Dict mapping model name to list of filters for that model
            - main_query_filters: Filters that reference multiple models or metrics (can't push down)
            - window_dim_filters_by_model: Dict mapping model name to filters that reference
              a window dimension on that model. These can't be pushed into CTE WHERE (window
              not yet evaluated). In the standard path they belong in the outer query; in the
              preagg path they are pushed into each model's sub-query (which has its own outer
              WHERE). Multi-model filters are keyed by the model that owns the window dim.
        """
        pushdown_filters = {model: [] for model in all_models}
        main_query_filters = []
        window_dim_filters: dict[str, list[str]] = {model: [] for model in all_models}

        # Flatten compound AND expressions into individual conjuncts so each
        # part can be classified independently.
        flat_parts: list[str] = []
        for filter_expr in filters:
            try:
                parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
                conjuncts = list(parsed.flatten() if isinstance(parsed, exp.And) else [parsed])
                flat_parts.extend(c.sql(dialect=self.dialect) for c in conjuncts)
            except Exception:
                flat_parts.append(filter_expr)

        for filter_expr in flat_parts:
            # Parse filter expression with SQLGlot
            try:
                parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
            except Exception:
                # If parsing fails, keep in main query to be safe
                main_query_filters.append(filter_expr)
                continue

            # Find all table references in the filter
            referenced_models = set()
            references_metric = False
            window_dim_models: set[str] = set()

            for column in parsed.find_all(exp.Column):
                table_name = column.table
                column_name = column.name

                if table_name:
                    # Remove _cte suffix if present
                    clean_name = table_name.replace("_cte", "")
                    if clean_name in all_models:
                        referenced_models.add(clean_name)

                        # Check if this column is a metric
                        model = self.graph.get_model(clean_name)
                        if model and model.get_metric(column_name):
                            references_metric = True

                        # Check if this column is a window dimension
                        if model:
                            dim = model.get_dimension(column_name)
                            if dim and getattr(dim, "window", None) is not None:
                                window_dim_models.add(clean_name)

            # Window-dim filters take priority: kept separate so callers can
            # handle them appropriately.  In the standard path they go to the
            # outer WHERE; in the preagg path they are pushed into each model's
            # sub-query (which has its own outer WHERE after window evaluation).
            # This check must come BEFORE the metric check because a filter
            # that references both a metric and a window dimension must NOT
            # land in main_query_filters/shared_filters: preagg CTEs do not
            # project window dimension columns.
            # For multi-model filters, route to each model that owns a window
            # dim column so the preagg path includes them in the correct
            # sub-queries (the recursive generate() call will join in any
            # additional models referenced by the filter).
            if window_dim_models:
                for wdm in window_dim_models:
                    window_dim_filters[wdm].append(filter_expr)
            # Filters that reference metrics must stay in main query (can't push down)
            # because metrics don't exist in CTEs (only _raw columns)
            elif references_metric:
                main_query_filters.append(filter_expr)
            # If filter references exactly one model and no metrics, push it down
            elif len(referenced_models) == 1:
                model = list(referenced_models)[0]
                pushdown_filters[model].append(filter_expr)
            else:
                # Filter references multiple models or no models - keep in main query
                main_query_filters.append(filter_expr)

        return pushdown_filters, main_query_filters, window_dim_filters

    def _extract_metric_filter_columns(self, metrics: list[str]) -> dict[str, set[str]]:
        """Extract columns referenced in metric-level filters and SQL expressions.

        Recursively extracts filter columns from:
        - Direct measure references (model.measure)
        - Graph-level metrics with filters
        - Ratio/derived metrics that reference measures with filters

        Args:
            metrics: List of metric references (e.g., ["orders.revenue", "bookings.gross_value"])

        Returns:
            Dict mapping model_name -> set of column names needed for metric filters and SQL expressions
        """
        columns_by_model: dict[str, set[str]] = {}

        def add_filter_columns(model_name: str, filters: list[str]):
            """Extract columns from filter expressions and add to columns_by_model."""
            if model_name not in columns_by_model:
                columns_by_model[model_name] = set()
            for f in filters:
                aliased_filter = f.replace("{model}", f"{model_name}_cte")
                try:
                    parsed = sqlglot.parse_one(aliased_filter, dialect=self.dialect)
                    for col in parsed.find_all(exp.Column):
                        if col.table and col.table.replace("_cte", "") == model_name:
                            columns_by_model[model_name].add(col.name)
                        elif not col.table:
                            columns_by_model[model_name].add(col.name)
                except Exception:
                    logging.debug("Failed to parse metric filter: %s", aliased_filter, exc_info=True)

        def add_sql_columns(sql_expr: str, default_model_name: str | None = None):
            """Extract column refs from SQL and track them per model."""
            try:
                parsed = sqlglot.parse_one(sql_expr, dialect=self.dialect)
                for col in parsed.find_all(exp.Column):
                    if col.table:
                        model_name = col.table.replace("_cte", "")
                        if model_name in self.graph.models:
                            if model_name not in columns_by_model:
                                columns_by_model[model_name] = set()
                            columns_by_model[model_name].add(col.name)
                    elif default_model_name:
                        if default_model_name not in columns_by_model:
                            columns_by_model[default_model_name] = set()
                        columns_by_model[default_model_name].add(col.name)
            except Exception:
                pass

        def extract_from_measure_ref(metric_ref: str):
            """Extract filter columns from a model.measure reference."""
            if "." not in metric_ref:
                return
            if metric_ref in self.graph.metrics:
                extract_from_metric(self.graph.metrics[metric_ref])
                return
            model_name, measure_name = metric_ref.split(".", 1)
            model = self.graph.get_model(model_name)
            if model:
                measure = model.get_metric(measure_name)
                if measure:
                    # Check the measure's own filters
                    if measure.filters:
                        add_filter_columns(model_name, measure.filters)
                    # If measure is a ratio/derived, recursively check dependencies
                    if measure.type == "ratio":
                        if measure.numerator:
                            extract_from_measure_ref(measure.numerator)
                        if measure.denominator:
                            extract_from_measure_ref(measure.denominator)
                    elif measure.type == "derived" or (not measure.type and not measure.agg and measure.sql):
                        # For derived metrics, also extract columns from the SQL expression itself
                        # This handles inline SQL like: COUNT(CASE WHEN {model}.status = 'approved' THEN 1 END)
                        if measure.sql:
                            aliased_sql = measure.sql.replace("{model}", f"{model_name}_cte")
                            add_sql_columns(aliased_sql, model_name)
                        # Also check dependencies
                        deps = measure.get_dependencies(self.graph, model_name)
                        for dep in deps:
                            if "." in dep:
                                extract_from_measure_ref(dep)
                            else:
                                local_dep = model.get_metric(dep)
                                if local_dep:
                                    extract_from_measure_ref(f"{model_name}.{dep}")
                                    continue
                                try:
                                    dep_metric = self.graph.get_metric(dep)
                                    extract_from_metric(dep_metric)
                                except KeyError:
                                    pass

        def extract_from_metric(metric):
            """Recursively extract filter columns from a metric and its dependencies."""
            # Extract from the metric's own filters
            if metric.filters:
                deps = metric.get_dependencies(self.graph)
                for dep in deps:
                    try:
                        dep_model_name, _ = self.graph.resolve_metric_reference(dep)
                    except KeyError:
                        dep_model_name = dep.split(".", 1)[0] if "." in dep else None
                    if dep_model_name:
                        add_filter_columns(dep_model_name, metric.filters)
                        break

            # For ratio metrics, check numerator and denominator
            if metric.type == "ratio":
                if metric.numerator:
                    extract_from_measure_ref(metric.numerator)
                if metric.denominator:
                    extract_from_measure_ref(metric.denominator)

            # For derived metrics, check all dependencies
            elif metric.type == "derived" or (not metric.type and not metric.agg and metric.sql):
                deps = metric.get_dependencies(self.graph)
                for dep in deps:
                    if "." in dep:
                        extract_from_measure_ref(dep)
                    else:
                        # Recursively check graph-level metric dependencies
                        try:
                            dep_metric = self.graph.get_metric(dep)
                            extract_from_metric(dep_metric)
                        except KeyError:
                            pass

            # Also parse metric SQL directly so inline aggregate expressions
            # contribute required columns (critical for graph-level metrics).
            if metric.sql:
                add_sql_columns(metric.sql)

        for metric_ref in metrics:
            try:
                model_name, metric = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                continue
            if model_name:
                # model.measure format - extract directly
                extract_from_measure_ref(metric_ref)
            else:
                # Graph-level metric - recursively extract
                extract_from_metric(metric)

        return columns_by_model

    def _find_needed_dimensions(
        self,
        model_name: str,
        dimensions: list[tuple[str, str | None]],
        filters: list[str] | None,
        order_by: list[str] | None,
        metric_filter_columns: set[str] | None = None,
    ) -> set[str]:
        """Find which dimensions from this model are actually needed.

        Args:
            model_name: Model to check
            dimensions: Parsed dimension references from query
            filters: Filter expressions
            order_by: Order by fields
            metric_filter_columns: Columns needed for metric-level filters

        Returns:
            Set of dimension names needed for this model
        """
        needed = set()

        # Dimensions explicitly in SELECT
        for dim_ref, _ in dimensions:
            if dim_ref.startswith(model_name + "."):
                dim_name = dim_ref.split(".")[1]
                needed.add(dim_name)

        # Dimensions referenced in filters
        if filters:
            for filter_expr in filters:
                try:
                    parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
                    for col in parsed.find_all(exp.Column):
                        if col.table and col.table.replace("_cte", "") == model_name:
                            needed.add(col.name)
                except Exception:
                    logging.debug("Failed to parse filter for column extraction: %s", filter_expr, exc_info=True)

        # Dimensions referenced in ORDER BY
        if order_by:
            for order_field in order_by:
                # Strip DESC/ASC
                field = order_field.replace(" DESC", "").replace(" ASC", "").strip()
                if "." in field:
                    model_part, dim_part = field.split(".", 1)
                    if model_part == model_name:
                        needed.add(dim_part)

        # Columns needed for metric-level filters
        if metric_filter_columns:
            needed.update(metric_filter_columns)

        return needed

    def _build_model_cte(
        self,
        model_name: str,
        dimensions: list[tuple[str, str | None]],
        metrics: list[str],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        all_models: set[str] | None = None,
        metric_filter_columns: set[str] | None = None,
        ungrouped: bool = False,
    ) -> str:
        """Build CTE SQL for a model with optional filter pushdown.

        Args:
            model_name: Name of the model
            dimensions: Parsed dimension references
            metrics: Metric references
            filters: Filters to push down into this CTE (optional)
            order_by: Order by fields (for determining needed dimensions)
            all_models: All models in query (for determining if joins needed)
            metric_filter_columns: Columns needed for metric-level filters
            ungrouped: Whether the query is returning raw ungrouped rows

        Returns:
            CTE SQL string
        """
        model = self.graph.get_model(model_name)
        all_models = all_models or {model_name}
        needs_keyed_joins = self._model_needs_keyed_join_columns(model_name, all_models)

        # Find which dimensions are actually needed
        needed_dimensions = self._find_needed_dimensions(
            model_name, dimensions, filters, order_by, metric_filter_columns
        )

        # Build SELECT columns
        select_cols = []

        # Track all columns added (not just join keys) to avoid duplicates
        columns_added = set()

        def add_passthrough_column(column: str) -> None:
            if column not in columns_added:
                select_cols.append(f"{self._quote_identifier(column)} AS {self._quote_alias(column)}")
                columns_added.add(column)

        include_primary_keys = needs_keyed_joins or ungrouped or bool(model.sql)
        if include_primary_keys:
            for pk_col in model.primary_key_columns:
                add_passthrough_column(pk_col)

        # Include local relationship keys if we're joining OR if they're explicitly requested as dimensions
        for relationship in model.relationships:
            if relationship.type == "many_to_one":
                # Handle multi-column foreign keys
                for fk in relationship.foreign_key_columns:
                    # Add FK if: (1) we're joining to this related model, OR (2) FK is requested as dimension
                    should_include = (needs_keyed_joins and relationship.name in all_models) or fk in needed_dimensions
                    if should_include and fk not in columns_added:
                        add_passthrough_column(fk)
                        # Mark FK as "needed" so it's not duplicated as a dimension
                        needed_dimensions.discard(fk)
            elif (
                needs_keyed_joins
                and relationship.name in all_models
                and relationship.type in ("one_to_one", "one_to_many")
            ):
                local_keys = relationship.primary_key_columns if relationship.primary_key else model.primary_key_columns
                for pk in local_keys:
                    add_passthrough_column(pk)

        # Check if other models have has_many/has_one pointing to this model
        if needs_keyed_joins:
            for other_model_name, other_model in self.graph.models.items():
                if other_model_name not in all_models:
                    continue
                for other_join in other_model.relationships:
                    if other_join.name == model_name and other_join.type in (
                        "one_to_one",
                        "one_to_many",
                    ):
                        # Other model expects this model to have a foreign key
                        # For has_many/has_one, foreign_key is the FK column in THIS model
                        for fk in other_join.foreign_key_columns:
                            add_passthrough_column(fk)
                    elif other_join.name == model_name and other_join.type == "many_to_one":
                        target_keys = (
                            other_join.primary_key_columns if other_join.primary_key else model.primary_key_columns
                        )
                        for pk in target_keys:
                            add_passthrough_column(pk)

            for other_model_name, other_model in self.graph.models.items():
                if other_model_name not in all_models:
                    continue
                for other_join in other_model.relationships:
                    if other_join.type != "many_to_many" or other_join.through != model_name:
                        continue
                    junction_self_fks, junction_related_fks = other_join.junction_key_columns()
                    for fk in (*junction_self_fks, *junction_related_fks):
                        if fk and fk not in columns_added:
                            select_cols.append(f"{self._quote_identifier(fk)} AS {self._quote_alias(fk)}")
                            columns_added.add(fk)

        # Determine table alias for {model} placeholder replacement
        # In CTEs, we're selecting from the raw table (or subquery AS t)
        model_table_alias = "t" if model.sql else ""

        def replace_model_placeholder(sql_expr: str) -> str:
            """Replace {model} placeholder with appropriate table reference."""
            if model_table_alias:
                return sql_expr.replace("{model}", model_table_alias)
            else:
                # No alias needed - just remove {model}.
                return sql_expr.replace("{model}.", "")

        # Add only needed dimension columns
        for dimension in model.dimensions:
            if dimension.name in needed_dimensions and dimension.name not in columns_added:
                # For time dimensions with granularity, apply DATE_TRUNC
                # Use window_sql_expr for CTE projection so window functions
                # (LEAD, LAG, etc.) are evaluated here.
                base_expr = dimension.window_sql_expr
                if dimension.type == "time" and dimension.granularity:
                    dim_sql = self._date_trunc(dimension.granularity, base_expr)
                else:
                    dim_sql = base_expr
                # Replace {model} placeholder with actual table reference
                dim_sql = replace_model_placeholder(dim_sql)
                select_cols.append(f"{dim_sql} AS {self._quote_alias(dimension.name)}")
                columns_added.add(dimension.name)

        # Then, add time dimensions with specific granularities
        for dim_ref, gran in dimensions:
            if not dim_ref.startswith(model_name + "."):
                continue

            dim_name = dim_ref.split(".")[1]
            dimension = model.get_dimension(dim_name)

            if not dimension:
                continue

            if gran and dimension.type == "time":
                # Apply time granularity (in addition to base column)
                dim_sql = self._date_trunc(gran, replace_model_placeholder(dimension.sql_expr))
                alias = f"{dim_name}__{gran}"
                if alias not in columns_added:
                    select_cols.append(f"{dim_sql} AS {self._quote_alias(alias)}")
                    columns_added.add(alias)

        # Add measure columns (raw, not aggregated in CTE)
        # Collect all measures needed for metrics
        measures_needed = set()
        extra_metric_sql_columns: set[str] = set()

        def collect_sql_columns_for_model(sql_expr: str) -> None:
            try:
                parsed = sqlglot.parse_one(sql_expr, dialect=self.dialect)
                for col in parsed.find_all(exp.Column):
                    if col.table and col.table.replace("_cte", "") != model_name:
                        continue
                    extra_metric_sql_columns.add(col.name)
            except Exception:
                logging.debug("Failed to parse metric SQL for columns: %s", sql_expr, exc_info=True)

        def collect_measures_from_metric(metric_ref: str, visited: set[str] | None = None):
            """Recursively collect measures needed from a metric.

            Handles:
            - Direct measure references: model.measure
            - Model-level derived measures that reference other measures
            - Unqualified references that need to be resolved within model context
            """
            if visited is None:
                visited = set()

            # Avoid infinite recursion
            if metric_ref in visited:
                return
            visited.add(metric_ref)

            try:
                ref_model_name, resolved_metric = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                ref_model_name = None
                resolved_metric = None

            if resolved_metric and ref_model_name is None:
                for dep in resolved_metric.get_dependencies(self.graph, model_name):
                    collect_measures_from_metric(dep, visited)
                if resolved_metric.sql and (
                    resolved_metric.agg or sql_has_aggregate(resolved_metric.sql, self.dialect)
                ):
                    collect_sql_columns_for_model(resolved_metric.sql)
                return

            if resolved_metric and ref_model_name:
                # It's a qualified reference (model.measure)
                if ref_model_name != model_name:
                    return
                measure_name = metric_ref.split(".", 1)[1]
                measure = resolved_metric
                if (
                    not measure.type
                    and not measure.agg
                    and measure.sql
                    and sql_has_aggregate(measure.sql, self.dialect)
                ):
                    collect_sql_columns_for_model(measure.sql)
                    return
                if measure.type in ("derived", "ratio") or (not measure.type and not measure.agg and measure.sql):
                    # Derived/ratio measure - get its dependencies
                    for dep in measure.get_dependencies(self.graph, ref_model_name):
                        collect_measures_from_metric(dep, visited)
                elif measure.agg:
                    # Simple aggregation measure - add it
                    measures_needed.add(measure_name)
            elif "." in metric_ref:
                return
            else:
                # Unqualified reference - could be:
                # 1. A graph-level metric
                # 2. A measure on the current model

                # First check if it's a measure on the current model
                measure = model.get_metric(metric_ref)
                if measure:
                    if (
                        not measure.type
                        and not measure.agg
                        and measure.sql
                        and sql_has_aggregate(measure.sql, self.dialect)
                    ):
                        collect_sql_columns_for_model(measure.sql)
                        return
                    if measure.type in ("derived", "ratio") or (not measure.type and not measure.agg and measure.sql):
                        # Derived/ratio measure - get its dependencies
                        for dep in measure.get_dependencies(self.graph, model_name):
                            collect_measures_from_metric(dep, visited)
                    elif measure.agg:
                        # Simple aggregation measure - add it
                        measures_needed.add(metric_ref)
                else:
                    # Try as graph-level metric
                    try:
                        metric = self.graph.get_metric(metric_ref)
                        if metric:
                            # Use auto dependency detection with graph for resolution
                            for dep in metric.get_dependencies(self.graph, model_name):
                                collect_measures_from_metric(dep, visited)
                    except KeyError:
                        pass

        for metric_ref in metrics:
            collect_measures_from_metric(metric_ref)

        all_metric_columns = set(metric_filter_columns or set()) | extra_metric_sql_columns

        # Add raw columns referenced by inline aggregate SQL (if they are not dimensions/measures)
        for col_name in all_metric_columns:
            if col_name in columns_added:
                continue
            dim = model.get_dimension(col_name)
            if dim:
                dim_sql = replace_model_placeholder(dim.window_sql_expr)
                select_cols.append(f"{dim_sql} AS {self._quote_alias(col_name)}")
                columns_added.add(col_name)
                continue
            if model.get_metric(col_name):
                continue
            raw_expr = f"{model_table_alias}.{col_name}" if model_table_alias else col_name
            select_cols.append(f"{raw_expr} AS {self._quote_alias(col_name)}")
            columns_added.add(col_name)

        # Also include measure columns referenced in metric_filter_columns (for derived metrics
        # with inline SQL aggregations like "SUM(quantity * unit_price) / COUNT(DISTINCT order_id)")
        for col_name in all_metric_columns:
            # Check if this column is a measure (not a dimension)
            measure = model.get_metric(col_name)
            if measure and measure.agg and col_name not in measures_needed:
                measures_needed.add(col_name)

        for measure_name in measures_needed:
            measure = model.get_metric(measure_name)
            if measure:
                # Build the base SQL expression for the measure
                if measure.agg == "count" and (not measure.sql or measure.sql == "*"):
                    base_sql = "1"
                elif measure.agg == "count_distinct" and not measure.sql:
                    pk_cols = model.primary_key_columns
                    if len(pk_cols) == 1:
                        base_sql = self._quote_identifier(pk_cols[0])
                    else:
                        # For composite keys, concatenate columns for uniqueness
                        base_sql = (
                            "CONCAT("
                            + ", '|', ".join(f"CAST({self._quote_identifier(c)} AS VARCHAR)" for c in pk_cols)
                            + ")"
                        )
                else:
                    base_sql = replace_model_placeholder(measure.sql_expr)

                # Apply measure filters if present (wrap in CASE WHEN)
                if measure.filters:
                    # Filters are SQL conditions like "{model}.field = 'value'"
                    # Replace {model} placeholder and combine into CASE WHEN
                    filter_conditions = []
                    for filter_str in measure.filters:
                        # Replace {model} with nothing since we're in the CTE selecting from raw table
                        filter_sql = filter_str.replace("{model}.", "").replace("{model}", "")
                        filter_conditions.append(filter_sql)

                    if filter_conditions:
                        filter_sql = " AND ".join(filter_conditions)
                        # For count measures, return 1 if condition met, else NULL
                        # COUNT counts non-NULL values, so we need NULL to exclude non-matching rows
                        if measure.agg == "count":
                            measure_sql = f"CASE WHEN {filter_sql} THEN 1 ELSE NULL END"
                        else:
                            measure_sql = f"CASE WHEN {filter_sql} THEN {base_sql} ELSE NULL END"
                    else:
                        measure_sql = base_sql
                else:
                    measure_sql = base_sql

                select_cols.append(f"{measure_sql} AS {self._quote_alias(f'{measure_name}_raw')}")

        # Build FROM clause
        from_clause = self._model_from_clause(model)

        # Build WHERE clause for pushed-down filters
        where_clause = ""
        if filters:
            # Process filters - replace model_cte references with direct column names using SQLGlot
            processed_filters = []
            for f in filters:
                try:
                    parsed = sqlglot.parse_one(f, dialect=self.dialect)
                    # Remove table qualifiers (model_name_cte. or model_name.)
                    for col in parsed.find_all(exp.Column):
                        if col.table:
                            clean_table = col.table.replace("_cte", "")
                            if clean_table == model_name:
                                col.set("table", None)
                    processed_filter = parsed.sql(dialect=self.dialect)
                    processed_filters.append(processed_filter)
                except Exception:
                    # If parsing fails, use original filter
                    processed_filters.append(f)

            where_clause = f"\n  WHERE {' AND '.join(processed_filters)}"

        # Build CTE
        select_str = ",\n    ".join(select_cols)
        cte_sql = (
            f"{self._quote_identifier(self._cte_name(model_name))} AS "
            f"(\n  SELECT\n    {select_str}\n  FROM {from_clause}{where_clause}\n)"
        )

        return cte_sql

    def _model_needs_keyed_join_columns(self, model_name: str, all_models: set[str]) -> bool:
        """Return whether this model needs columns for non-cross joins in this query."""
        if len(all_models) <= 1:
            return False

        model = self.graph.get_model(model_name)
        for relationship in model.relationships:
            if relationship.name in all_models and relationship.type != "cross":
                return True

        for other_model_name, other_model in self.graph.models.items():
            if other_model_name not in all_models:
                continue
            for relationship in other_model.relationships:
                if relationship.name == model_name and relationship.type != "cross":
                    return True
                if (
                    relationship.type == "many_to_many"
                    and relationship.through == model_name
                    and relationship.name in all_models
                ):
                    return True

        return False

    def _has_fanout_joins(self, base_model_name: str, other_models: list[str]) -> dict[str, bool]:
        """Determine which models need symmetric aggregates due to fan-out.

        When one-to-many joins exist from the base model, measures from
        the base model need symmetric aggregates to prevent double-counting.

        Args:
            base_model_name: Base model name
            other_models: Other models in the query

        Returns:
            Dict mapping model names to whether they need symmetric aggregates
        """
        needs_symmetric = {}

        # Check if there are any one-to-many relationships
        one_to_many_count = 0
        many_to_one_models = []

        for other_model in other_models:
            try:
                join_path = self.graph.find_relationship_path(base_model_name, other_model)
                if not join_path:
                    continue
                # Check all hops: any one_to_many in the path creates fan-out
                has_fanout = any(hop.relationship == "one_to_many" for hop in join_path)
                if has_fanout:
                    one_to_many_count += 1
                elif join_path[0].relationship == "many_to_one":
                    many_to_one_models.append(other_model)
            except (ValueError, KeyError):
                pass

        # Base model needs symmetric aggregates if there are any one-to-many joins
        needs_symmetric[base_model_name] = one_to_many_count > 0

        # Models on the "many" side of a many-to-one relationship also need symmetric
        # aggregation if they're being joined (because from their perspective,
        # they're creating fan-out for the "one" side)
        for other_model in other_models:
            if other_model in many_to_one_models:
                # Check if the "one" side (base) has metrics - if so, it needs symmetric agg
                # But we're checking from the perspective of this model, so mark False
                needs_symmetric[other_model] = False
            else:
                needs_symmetric[other_model] = False

        return needs_symmetric

    def _explicit_join_type_for_path(self, join_path) -> str | None:
        """Return an adapter-provided SQL join type for a join path step."""
        relation = None
        reverse = False

        try:
            from_model = self.graph.get_model(join_path.from_model)
            relation = next((rel for rel in from_model.relationships if rel.name == join_path.to_model), None)
        except KeyError:
            relation = None

        if relation is None:
            try:
                to_model = self.graph.get_model(join_path.to_model)
                relation = next((rel for rel in to_model.relationships if rel.name == join_path.from_model), None)
                reverse = relation is not None
            except KeyError:
                relation = None

        if not relation or not relation.metadata or "bsl_how" not in relation.metadata:
            return None

        how = str(relation.metadata["bsl_how"]).lower()
        if reverse and how == "left":
            how = "right"
        elif reverse and how == "right":
            how = "left"

        return {
            "inner": "inner",
            "left": "left",
            "right": "right",
            "outer": "full",
            "full": "full",
            "full_outer": "full",
        }.get(how)

    def _needs_preaggregation_for_fanout(self, metrics: list[str], dimensions: list[str]) -> bool:
        """Determine if pre-aggregation is needed to avoid fan-out.

        Pre-aggregation is needed when:
        1. Metrics come from multiple different models
        2. Those models are at different levels in the join chain
        3. A join between them would cause one model's metrics to be over-counted

        For example: employees.total_salary + departments.total_budget by companies.name
        The join path is: companies -> departments -> employees
        When employees join to departments, each department row is replicated per employee,
        causing department budgets to be summed multiple times.

        Args:
            metrics: List of metric references (e.g., ["employees.total_salary", "departments.total_budget"])
            dimensions: List of dimension references (e.g., ["companies.name"])

        Returns:
            True if pre-aggregation is needed
        """
        if not metrics or len(metrics) < 2:
            return False

        # Get unique metric models
        metric_models = set()
        for metric_ref in metrics:
            try:
                model_name, _ = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                model_name = None
            if model_name:
                metric_models.add(model_name)

        if len(metric_models) < 2:
            return False

        # Check if any pair of metric models would cause fan-out
        # Fan-out occurs when model A joins to model B via a path that includes
        # a many_to_one relationship from B's perspective (one_to_many from A's)
        metric_model_list = list(metric_models)
        for i, model_a in enumerate(metric_model_list):
            for model_b in metric_model_list[i + 1 :]:
                try:
                    # Check path from A to B
                    join_path = self.graph.find_relationship_path(model_a, model_b)
                    if join_path:
                        # If any hop is many_to_one (from A's perspective), model_a metrics
                        # would be replicated when joining to model_b
                        for jp in join_path:
                            if jp.relationship == "many_to_one":
                                # model_a is on the "many" side, so its rows fan out
                                # when we aggregate model_b metrics
                                return True

                    # Check reverse path
                    join_path_reverse = self.graph.find_relationship_path(model_b, model_a)
                    if join_path_reverse:
                        for jp in join_path_reverse:
                            if jp.relationship == "many_to_one":
                                return True

                except (ValueError, KeyError):
                    pass

        return False

    def _generate_with_preaggregation(
        self,
        metrics: list[str],
        dimensions: list[str],
        filters: list[str] | None = None,
        segments: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
        aliases: dict[str, str] | None = None,
        use_preaggregations: bool = False,
    ) -> str:
        """Generate SQL using pre-aggregation to avoid fan-out.

        This generates separate queries for each metric model, pre-aggregated
        to the dimension grain, then joins them together.

        Args:
            metrics: List of metric references
            dimensions: List of dimension references
            filters: List of filter expressions
            segments: List of segment references
            order_by: List of fields to order by
            limit: Maximum number of rows
            offset: Number of rows to skip
            aliases: Custom aliases for fields
            use_preaggregations: Try materialized pre-aggregation routing for each child query

        Returns:
            SQL query string
        """
        aliases = aliases or {}
        parsed_dims = self._parse_dimension_refs(dimensions)

        # Group metrics by their model
        metrics_by_model: dict[str, list[str]] = {}
        for metric_ref in metrics:
            try:
                model_name, _ = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                model_name = None
            if model_name:
                if model_name not in metrics_by_model:
                    metrics_by_model[model_name] = []
                metrics_by_model[model_name].append(metric_ref)

        if len(metrics_by_model) < 2:
            # Shouldn't happen, but fall back to regular generation.
            # Segments already resolved into filters by caller, pass None.
            return self.generate(
                metrics=metrics,
                dimensions=dimensions,
                filters=filters,
                segments=None,
                order_by=order_by,
                limit=limit,
                offset=offset,
                aliases=aliases,
                use_preaggregations=use_preaggregations,
            )

        # Resolve segments to SQL filters
        segment_filters = self._resolve_segments(segments or [])
        all_filters = (filters or []) + segment_filters

        # Partition filters by model so sub-queries only get relevant filters.
        # Cross-model filters (referencing models outside the sub-query) would
        # produce invalid SQL referencing CTEs that don't exist.
        all_model_names = set(metrics_by_model.keys())
        pushdown_by_model, shared_filters, window_dim_filters = self._classify_filters_for_pushdown(
            all_filters, all_model_names
        )

        # Generate a pre-aggregated CTE for each metric model
        preagg_ctes = []
        cte_names = []

        for model_name, model_metrics in metrics_by_model.items():
            cte_name = f"{model_name}_preagg"
            cte_names.append(cte_name)

            # Pass pushdown filters plus any window-dim filters for this model.
            # Window-dim filters are pushed into the model's sub-query (not the
            # outer preagg join) so the recursive generate() handles them in its
            # own outer WHERE, preserving the requested dimension grain.
            model_filters = pushdown_by_model.get(model_name, []) + window_dim_filters.get(model_name, [])

            # Generate sub-query for this model's metrics at the dimension grain
            # We call generate() recursively but it won't trigger pre-aggregation
            # again because each sub-query has metrics from only one model
            sub_query = self.generate(
                metrics=model_metrics,
                dimensions=dimensions,
                filters=model_filters,
                segments=None,  # Already resolved
                order_by=None,
                limit=None,
                offset=None,
                aliases=aliases,
                use_preaggregations=use_preaggregations,
            )

            # Remove the instrumentation comment from sub-query
            sub_query_lines = sub_query.split("\n")
            sub_query_clean = "\n".join(
                line for line in sub_query_lines if not line.strip().startswith("-- sidemantic:")
            )

            preagg_ctes.append(f"{cte_name} AS (\n{sub_query_clean}\n)")

        # Build the final SELECT that joins all pre-aggregated CTEs
        select_exprs = []
        output_names: dict[str, str] = {}

        def register_output_name(output_name: str, *refs: str) -> None:
            for ref in refs:
                if ref:
                    output_names[ref] = output_name

        def dimension_output_name(dim_ref: str, dim_name: str, gran: str | None) -> str:
            if gran:
                full_ref = f"{dim_ref}__{gran}"
                default = f"{dim_name}__{gran}"
            else:
                full_ref = dim_ref
                default = dim_name

            canonical_ref = full_ref
            return aliases.get(full_ref) or aliases.get(canonical_ref) or default

        def metric_source_name(metric_ref: str, metric_name: str) -> str:
            return aliases.get(metric_ref) or aliases.get(f"{metric_ref.split('.')[0]}.{metric_name}") or metric_name

        # Add dimensions - use COALESCE across all CTEs
        for dim_ref, gran in parsed_dims:
            dim_name = dim_ref.split(".")[1] if "." in dim_ref else dim_ref
            col_name = f"{dim_name}__{gran}" if gran else dim_name
            output_name = dimension_output_name(dim_ref, dim_name, gran)
            full_ref = f"{dim_ref}__{gran}" if gran else dim_ref
            canonical_ref = full_ref
            register_output_name(output_name, full_ref, canonical_ref, dim_name, col_name, output_name)

            # Build COALESCE expression
            quoted_source = self._quote_identifier(output_name)
            coalesce_parts = [f"{cte}.{quoted_source}" for cte in cte_names]
            select_exprs.append(f"COALESCE({', '.join(coalesce_parts)}) AS {self._quote_alias(output_name)}")

        # Check for metric name collisions across models
        metric_name_counts: dict[str, int] = {}
        for model_metrics in metrics_by_model.values():
            for metric_ref in model_metrics:
                metric_name = metric_ref.split(".", 1)[1] if "." in metric_ref else metric_ref
                metric_name_counts[metric_name] = metric_name_counts.get(metric_name, 0) + 1

        # Add metrics from each CTE
        for model_name, model_metrics in metrics_by_model.items():
            cte_name = f"{model_name}_preagg"
            for metric_ref in model_metrics:
                metric_name = metric_ref.split(".", 1)[1] if "." in metric_ref else metric_ref
                source_name = metric_source_name(metric_ref, metric_name)
                # Check for custom alias first
                if metric_ref in aliases:
                    alias = aliases[metric_ref]
                elif metric_name_counts.get(metric_name, 1) > 1:
                    # Collision - prefix with model name
                    alias = f"{model_name}_{metric_name}"
                else:
                    alias = metric_name
                register_output_name(alias, metric_ref, f"{model_name}.{metric_name}", metric_name, alias)
                select_exprs.append(f"{cte_name}.{self._quote_identifier(source_name)} AS {self._quote_alias(alias)}")

        # Build FROM clause with FULL OUTER JOINs (or CROSS JOIN if no dimensions)
        # Start with first CTE
        from_clause = cte_names[0]

        # Join remaining CTEs
        join_clauses = []
        for cte_name in cte_names[1:]:
            if not parsed_dims:
                # No dimensions - use CROSS JOIN (each CTE returns single row)
                join_clauses.append(f"CROSS JOIN {cte_name}")
            else:
                # Build join condition on all dimension columns
                join_conditions = []
                for dim_ref, gran in parsed_dims:
                    dim_name = dim_ref.split(".")[1] if "." in dim_ref else dim_ref
                    col_name = dimension_output_name(dim_ref, dim_name, gran)
                    # NULL-safe equality that works for all column types
                    lhs = exp.column(col_name, table=cte_names[0])
                    rhs = exp.column(col_name, table=cte_name)
                    join_conditions.append(exp.NullSafeEQ(this=lhs, expression=rhs).sql(dialect=self.dialect))

                join_clause = " AND ".join(join_conditions)
                join_clauses.append(f"FULL OUTER JOIN {cte_name} ON {join_clause}")

        # Combine into final query
        select_str = ",\n  ".join(select_exprs)
        from_str = from_clause + "\n" + "\n".join(join_clauses)

        final_query = f"SELECT\n  {select_str}\nFROM {from_str}"

        # Apply shared filters (cross-model or metric-level) on the outer query.
        # Rewrite table references from model/model_cte to model_preagg using
        # sqlglot AST rewriting so we don't accidentally mangle column names
        # that happen to contain a model name as a substring.
        if shared_filters:
            preagg_table_map = {}
            for model_name in metrics_by_model:
                preagg_table_map[model_name] = f"{model_name}_preagg"
                preagg_table_map[f"{model_name}_cte"] = f"{model_name}_preagg"

            rewritten = []
            for f in shared_filters:
                try:
                    parsed = sqlglot.parse_one(f, dialect=self.dialect)
                    for col in parsed.find_all(exp.Column):
                        if col.table and col.table in preagg_table_map:
                            col.set("table", exp.to_identifier(preagg_table_map[col.table]))
                    rewritten.append(parsed.sql(dialect=self.dialect))
                except Exception:
                    # Parsing failed, fall back to the raw filter expression.
                    # This is best-effort: the filter may reference CTE names
                    # that don't exist on the outer query, but that will surface
                    # as a clear SQL error rather than silently dropping the filter.
                    rewritten.append(f)
            final_query += f"\nWHERE {' AND '.join(rewritten)}"

        # Add ORDER BY
        if order_by:
            order_clauses = []
            for field in order_by:
                parts = field.rsplit(" ", 1)
                direction = ""
                field_ref = field
                if len(parts) == 2 and parts[1].upper() in {"ASC", "DESC"}:
                    field_ref = parts[0]
                    direction = f" {parts[1].upper()}"

                field_name = field_ref.split(".", 1)[1] if "." in field_ref else field_ref
                output_name = output_names.get(field_ref) or output_names.get(field_name) or field_ref
                order_clauses.append(f"{self._quote_alias(output_name)}{direction}")
            final_query += f"\nORDER BY {', '.join(order_clauses)}"

        # Add LIMIT and OFFSET
        if limit is not None:
            final_query += f"\nLIMIT {limit}"
        if offset is not None:
            final_query += f"\nOFFSET {offset}"

        # Combine CTEs and main query
        cte_str = "WITH " + ",\n".join(preagg_ctes)
        full_sql = cte_str + "\n" + final_query

        # Add instrumentation comment
        all_models = list(metrics_by_model.keys())
        instrumentation = self._generate_instrumentation_comment(
            models=all_models, metrics=metrics, dimensions=dimensions, used_preagg=False
        )
        full_sql = full_sql + "\n" + instrumentation

        return full_sql

    def _add_join_paths_to_query(
        self,
        query,
        base_model_name: str,
        other_models: list[str],
        models_with_filters: set[str],
    ):
        """Add the same multi-hop joins used by the main grouped query."""
        if not other_models:
            return query

        joined_models = {base_model_name}
        for other_model in other_models:
            join_path = self.graph.find_relationship_path(base_model_name, other_model)
            if not join_path:
                continue

            for jp in join_path:
                if jp.to_model in joined_models:
                    continue

                right_table = self._quote_identifier(self._cte_name(jp.to_model))
                if jp.relationship == "cross":
                    query = query.join(right_table, join_type="cross")
                    joined_models.add(jp.to_model)
                    continue

                if jp.custom_condition:
                    join_cond = self._custom_join_condition(jp)
                else:
                    if len(jp.from_columns) != len(jp.to_columns):
                        raise ValueError(
                            f"Join between {jp.from_model} and {jp.to_model} has mismatched key columns: "
                            f"from_columns has {len(jp.from_columns)}, to_columns has {len(jp.to_columns)}"
                        )

                    join_conditions = [
                        self._cte_ref(jp.from_model, fk) + " = " + self._cte_ref(jp.to_model, pk)
                        for fk, pk in zip(jp.from_columns, jp.to_columns)
                    ]
                    join_cond = " AND ".join(join_conditions)
                join_type = self._explicit_join_type_for_path(jp)
                if join_type is None:
                    join_type = "inner" if jp.to_model in models_with_filters else "left"
                query = query.join(right_table, on=join_cond, join_type=join_type)
                joined_models.add(jp.to_model)

        return query

    def _split_where_having_filters(self, filters: list[str], model_names: list[str]) -> tuple[list[str], list[str]]:
        """Split query-level filters into row filters and aggregate filters."""
        import re

        where_filters = []
        having_filters = []

        for filter_expr in filters:
            references_metric = False

            for model_name in model_names:
                model_obj = self.graph.get_model(model_name)
                pattern = f"{model_name}\\.([a-zA-Z_][a-zA-Z0-9_]*)"
                matches = re.findall(pattern, filter_expr)
                for field_name in matches:
                    if model_obj.get_metric(field_name):
                        references_metric = True
                        break
                if references_metric:
                    break

            references_window_dim = False
            if references_metric:
                for model_name in model_names:
                    model_obj = self.graph.get_model(model_name)
                    if not model_obj:
                        continue
                    for field_name in re.findall(f"{model_name}\\.([a-zA-Z_][a-zA-Z0-9_]*)", filter_expr):
                        dim = model_obj.get_dimension(field_name)
                        if dim and getattr(dim, "window", None) is not None:
                            references_window_dim = True
                            break
                    if references_window_dim:
                        break

            if references_metric and not references_window_dim:
                having_filters.append(filter_expr)
            else:
                where_filters.append(filter_expr)

        return where_filters, having_filters

    def _add_where_filters_to_query(self, query, where_filters: list[str], model_names: list[str]):
        """Add row-level filters with CTE-qualified field references."""
        import re

        for filter_expr in where_filters:
            parsed_filter = filter_expr
            for model_name in model_names:
                model_obj = self.graph.get_model(model_name)

                parts = []
                in_quotes = False
                current = ""

                for char in parsed_filter:
                    if char == "'":
                        if current:
                            parts.append((current, in_quotes))
                            current = ""
                        in_quotes = not in_quotes
                        parts.append(("'", False))
                    else:
                        current += char
                if current:
                    parts.append((current, in_quotes))

                pattern = f"{model_name}\\.([a-zA-Z_][a-zA-Z0-9_]*)"

                def replace_field(match):
                    field_name = match.group(1)
                    if model_obj.get_metric(field_name):
                        return self._cte_ref(model_name, f"{field_name}_raw")
                    return self._cte_ref(model_name, field_name)

                result_parts = []
                for part, is_quoted in parts:
                    if is_quoted or part == "'":
                        result_parts.append(part)
                    else:
                        result_parts.append(re.sub(pattern, replace_field, part))

                parsed_filter = "".join(result_parts)

            query = query.where(parsed_filter)

        return query

    def _build_main_select(
        self,
        base_model_name: str,
        other_models: list[str],
        parsed_dims: list[tuple[str, str | None]],
        metrics: list[str],
        filters: list[str] | None,
        models_with_filters: set[str],
        order_by: list[str] | None,
        limit: int | None,
        offset: int | None = None,
        ungrouped: bool = False,
        aliases: dict[str, str] | None = None,
    ) -> str:
        """Build main SELECT using SQLGlot builder API.

        Args:
            base_model_name: Base model name
            other_models: Other models to join
            parsed_dims: Parsed dimensions with granularities
            metrics: Metric references
            filters: Filter expressions
            models_with_filters: Set of models that have filters (for INNER JOIN)
            order_by: Order by fields
            limit: Row limit
            offset: Row offset
            ungrouped: If True, return raw rows without aggregation
            aliases: Custom aliases for fields (dict mapping field reference to alias)

        Returns:
            SQL SELECT statement
        """
        aliases = aliases or {}
        # Detect if symmetric aggregates are needed
        symmetric_agg_needed = self._has_fanout_joins(base_model_name, other_models)

        dimension_models = {dim_ref.split(".")[0] for dim_ref, _ in parsed_dims if "." in dim_ref}
        metric_models = set()
        for metric_ref in metrics:
            try:
                metric_model_name, _ = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                metric_model_name = None
            if metric_model_name:
                metric_models.add(metric_model_name)
        for metric_model in metric_models:
            for dimension_model in dimension_models:
                if metric_model == dimension_model:
                    continue
                try:
                    join_path = self.graph.find_relationship_path(metric_model, dimension_model)
                except ValueError:
                    continue
                if any(jp.relationship == "one_to_many" for jp in join_path):
                    symmetric_agg_needed[metric_model] = True
                    break

        # Check for dimension/metric name collisions across models
        # If there are collisions, prefix with model name
        field_names = {}  # field_name -> list of model names
        for dim_ref, gran in parsed_dims:
            model_name, dim_name = dim_ref.split(".")
            field_key = f"{dim_name}__{gran}" if gran else dim_name
            if field_key not in field_names:
                field_names[field_key] = []
            field_names[field_key].append(model_name)

        for metric_ref in metrics:
            try:
                model_name, resolved_metric = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                model_name = None
                resolved_metric = None
            if model_name:
                measure_name = metric_ref.split(".", 1)[1]
                if measure_name not in field_names:
                    field_names[measure_name] = []
                field_names[measure_name].append(model_name)
            elif resolved_metric:
                if resolved_metric.name not in field_names:
                    field_names[resolved_metric.name] = []
                field_names[resolved_metric.name].append("")

        # Determine which fields have collisions
        has_collision = {name: len(models) > 1 for name, models in field_names.items()}

        # Build SELECT columns
        select_exprs = []
        output_aliases: dict[str, str] = {}

        # Add dimensions
        for dim_ref, gran in parsed_dims:
            model_name, dim_name = dim_ref.split(".")
            cte_col_name = f"{dim_name}__{gran}" if gran else dim_name

            # Check for custom alias first
            full_ref = f"{model_name}.{dim_name}__{gran}" if gran else dim_ref
            base_alias = f"{dim_name}__{gran}" if gran else dim_name
            if full_ref in aliases:
                alias = aliases[full_ref]
            else:
                # Generate alias (with model prefix if collision)
                if has_collision.get(base_alias, False):
                    alias = f"{model_name}_{base_alias}"
                else:
                    alias = base_alias

            select_exprs.append(f"{self._cte_ref(model_name, cte_col_name)} AS {self._quote_alias(alias)}")
            output_aliases[full_ref] = alias
            output_aliases[dim_ref] = alias
            output_aliases[base_alias] = alias

        previous_bsl_all_context = getattr(self, "_bsl_all_query_context", None)
        self._bsl_all_query_context = {
            "base_model_name": base_model_name,
            "other_models": other_models,
            "filters": filters or [],
            "models_with_filters": models_with_filters,
        }

        # Add metrics
        for metric_ref in metrics:
            try:
                resolved_model_name, resolved_metric = self.graph.resolve_metric_reference(metric_ref)
            except KeyError:
                resolved_model_name = None
                resolved_metric = None

            if resolved_metric and resolved_model_name is None:
                metric_expr = self._build_metric_sql(resolved_metric)
                metric_expr = self._wrap_with_fill_nulls(metric_expr, resolved_metric)
                alias = aliases.get(metric_ref, resolved_metric.name)
                select_exprs.append(f"{metric_expr} AS {self._quote_alias(alias)}")
                output_aliases[metric_ref] = alias
                output_aliases[resolved_metric.name] = alias
            elif resolved_metric and resolved_model_name:
                # It's a measure reference (model.measure)
                model_name = resolved_model_name
                measure_name = metric_ref.split(".", 1)[1]
                measure = resolved_metric

                if measure:
                    # Check for custom alias first
                    if metric_ref in aliases:
                        alias = aliases[metric_ref]
                    else:
                        # Add model prefix if there's a collision
                        if has_collision.get(measure_name, False):
                            alias = f"{model_name}_{measure_name}"
                        else:
                            alias = measure_name

                    output_aliases[metric_ref] = alias
                    output_aliases[measure_name] = alias

                    # Complex metric types (derived, ratio) can be built inline
                    # Note: cumulative, time_comparison, conversion are handled via special query generators
                    # and won't appear in this code path
                    # Also handle "expression metrics" - metrics with inline aggregations like SUM(x)/SUM(y)
                    is_expression_metric = not measure.type and not measure.agg and measure.sql
                    if measure.type in ["derived", "ratio"] or is_expression_metric:
                        # Use complex metric builder
                        metric_expr = self._build_metric_sql(measure, model_name)
                        metric_expr = self._wrap_with_fill_nulls(metric_expr, measure)
                        select_exprs.append(f"{metric_expr} AS {self._quote_alias(alias)}")
                    elif not measure.agg:
                        # Unknown metric type that needs special handling
                        raise ValueError(
                            f"Metric '{measure.name}' with type '{measure.type}' cannot be queried directly. "
                            f"Use generate() instead of _build_main_select() for this metric type."
                        )
                    elif ungrouped:
                        # For ungrouped queries, select raw column without aggregation
                        select_exprs.append(
                            f"{self._cte_ref(model_name, f'{measure_name}_raw')} AS {self._quote_alias(alias)}"
                        )
                    else:
                        # Simple aggregation measures
                        # Check if this model needs symmetric aggregates
                        if symmetric_agg_needed.get(model_name, False):
                            # Use symmetric aggregates to prevent double-counting
                            # Get primary key for this model
                            model_obj = self.graph.get_model(model_name)
                            pk_cols = model_obj.primary_key_columns
                            # For composite keys, concatenate columns for hashing
                            if len(pk_cols) == 1:
                                pk = self._cte_ref(model_name, pk_cols[0])
                            else:
                                pk = (
                                    "CONCAT("
                                    + ", '|', ".join(
                                        f"CAST({self._cte_ref(model_name, c)} AS VARCHAR)" for c in pk_cols
                                    )
                                    + ")"
                                )

                            agg_expr = build_symmetric_aggregate_sql(
                                measure_expr=self._cte_ref(model_name, f"{measure_name}_raw"),
                                primary_key=pk,
                                agg_type=measure.agg,
                                dialect=self.dialect,
                            )
                        else:
                            # Use helper that applies metric-level filters via CASE WHEN
                            # This ensures each metric's filter only affects that metric
                            agg_expr = self._build_measure_aggregation_sql(model_name, measure)

                        select_exprs.append(f"{agg_expr} AS {self._quote_alias(alias)}")
                else:
                    # Try as metric
                    metric = self.graph.get_metric(metric_ref)
                    if metric:
                        metric_expr = self._build_metric_sql(metric)
                        metric_expr = self._wrap_with_fill_nulls(metric_expr, metric)
                        select_exprs.append(f"{metric_expr} AS {self._quote_alias(metric.name)}")
                        output_aliases[metric_ref] = metric.name
                        output_aliases[metric.name] = metric.name
            else:
                # It's a metric reference (just metric name)
                metric = self.graph.get_metric(metric_ref)
                if metric:
                    metric_expr = self._build_metric_sql(metric)
                    metric_expr = self._wrap_with_fill_nulls(metric_expr, metric)
                    select_exprs.append(f"{metric_expr} AS {self._quote_alias(metric.name)}")
                    output_aliases[metric_ref] = metric.name
                    output_aliases[metric.name] = metric.name
                else:
                    raise ValueError(f"Metric {metric_ref} not found")

        if previous_bsl_all_context is None:
            delattr(self, "_bsl_all_query_context")
        else:
            self._bsl_all_query_context = previous_bsl_all_context

        # Build query using builder API
        query = select(*select_exprs).from_(self._quote_identifier(self._cte_name(base_model_name)))

        # Add joins (supports multi-hop)
        query = self._add_join_paths_to_query(query, base_model_name, other_models, models_with_filters)

        # Separate filters into WHERE (dimension/row-level) and HAVING (metric/aggregation-level)
        # Note: metric-level filters (Metric.filters) are applied via CASE WHEN inside each
        # metric's aggregation, NOT in the WHERE clause. This ensures each metric's filter
        # only affects that specific metric, not all metrics in the query.
        where_filters, having_filters = self._split_where_having_filters(
            filters or [], [base_model_name] + other_models
        )
        import re

        # Add WHERE clause (dimension filters only - metric-level filters are in CASE WHEN)
        query = self._add_where_filters_to_query(query, where_filters, [base_model_name] + other_models)

        # Add GROUP BY (all dimensions by position)
        # Skip GROUP BY for ungrouped queries
        if parsed_dims and not ungrouped:
            group_by_positions = list(range(1, len(parsed_dims) + 1))
            query = query.group_by(*group_by_positions)

        # Add HAVING clause (metric filters applied after aggregation)
        if having_filters:
            for filter_expr in having_filters:
                # Replace model.metric_name with metric_name (the aggregated column alias)
                parsed_having = filter_expr
                for model_name in [base_model_name] + other_models:
                    model_obj = self.graph.get_model(model_name)
                    # Replace model.field with just field (aggregated column name)
                    pattern = f"{model_name}\\.([a-zA-Z_][a-zA-Z0-9_]*)"

                    def replace_metric_ref(match):
                        field_name = match.group(1)
                        # Just use the metric name (it's the SELECT alias)
                        return field_name

                    parsed_having = re.sub(pattern, replace_metric_ref, parsed_having)

                query = query.having(parsed_having)

        # Add ORDER BY
        if order_by:
            order_by_aliases = []
            for field in order_by:
                parts = field.rsplit(" ", 1)
                direction = ""
                field_ref = field
                if len(parts) == 2 and parts[1].upper() in {"ASC", "DESC"}:
                    field_ref = parts[0]
                    direction = f" {parts[1].upper()}"

                if field_ref in output_aliases:
                    field_alias = self._quote_alias(output_aliases[field_ref])
                elif "." in field_ref:
                    field_alias = field_ref.split(".", 1)[1]
                else:
                    field_alias = field_ref
                order_by_aliases.append(f"{field_alias}{direction}")
            query = query.order_by(*order_by_aliases)

        # Add LIMIT and OFFSET
        if limit is not None:
            query = query.limit(limit)
        if offset is not None:
            query = query.offset(offset)

        return query.sql(dialect=self.dialect, pretty=True)

    def _calculate_lag_offset(self, comparison_type: str | None, time_granularity: str | None) -> int:
        """Calculate LAG offset based on comparison type and time dimension granularity.

        LAG is row-based, so offsets for sub-period granularities are approximate:
        MoM at day grain uses 30 (months vary 28-31), YoY at day grain uses 365
        (ignores leap years), QoQ at day grain uses 90 (quarters vary 90-92).
        For exact period comparisons, use matching granularity (e.g., monthly grain
        for MoM) or a date-based self-join approach.

        Args:
            comparison_type: Type of comparison (yoy, mom, wow, dod, qoq)
            time_granularity: Time dimension granularity (day, week, month, quarter, year)
                             None means use the default offset for the comparison type

        Returns:
            Number of rows to offset for LAG function
        """
        if not comparison_type:
            return 1

        # When no explicit granularity is specified, use a sensible default
        # This assumes the data is at a matching granularity (e.g., monthly data for YoY = 12 rows)
        if not time_granularity:
            default_offsets = {
                "dod": 1,
                "wow": 1,
                "mom": 1,
                "qoq": 1,
                "yoy": 12,  # Assume monthly data for YoY
                "prior_period": 1,
            }
            return default_offsets.get(comparison_type, 1)

        # For simple comparison types, the offset depends on the granularity
        # e.g., YoY with monthly data = 12 months back, YoY with daily data = 365 days back
        offset_map = {
            # Day-over-day is always 1 period back
            "dod": {"day": 1, "week": 1, "month": 1, "quarter": 1, "year": 1},
            # Week-over-week
            "wow": {"day": 7, "week": 1, "month": 1, "quarter": 1, "year": 1},
            # Month-over-month
            "mom": {"day": 30, "week": 4, "month": 1, "quarter": 1, "year": 1},
            # Quarter-over-quarter
            "qoq": {"day": 90, "week": 13, "month": 3, "quarter": 1, "year": 1},
            # Year-over-year
            "yoy": {"day": 365, "week": 52, "month": 12, "quarter": 4, "year": 1},
            # Prior period (default to 1)
            "prior_period": {"day": 1, "week": 1, "month": 1, "quarter": 1, "year": 1},
        }

        if comparison_type in offset_map:
            return offset_map[comparison_type].get(time_granularity, 1)

        return 1

    def _wrap_with_fill_nulls(self, sql_expr: str, metric) -> str:
        """Wrap SQL expression with COALESCE if fill_nulls_with is specified.

        Args:
            sql_expr: Base SQL expression
            metric: Metric object

        Returns:
            Wrapped SQL expression
        """
        if metric.fill_nulls_with is not None:
            # Quote string values
            if isinstance(metric.fill_nulls_with, str):
                fill_value = f"'{metric.fill_nulls_with}'"
            else:
                fill_value = str(metric.fill_nulls_with)
            return f"COALESCE({sql_expr}, {fill_value})"
        return sql_expr

    def _rewrite_model_refs_to_ctes(self, sql_expr: str) -> str:
        """Rewrite qualified model refs (model.col) to CTE refs (model_cte.col)."""
        try:
            parsed = sqlglot.parse_one(sql_expr, dialect=self.dialect)
            for col in parsed.find_all(exp.Column):
                if not col.table:
                    continue
                model_name = col.table.replace("_cte", "")
                if model_name in self.graph.models:
                    cte_name = self._cte_name(model_name)
                    col.set("table", exp.to_identifier(cte_name, quoted=not self._is_simple_identifier(cte_name)))
            return parsed.sql(dialect=self.dialect)
        except Exception:
            rewritten = sql_expr
            for model_name in self.graph.models:
                rewritten = rewritten.replace(
                    f"{model_name}.",
                    f"{self._quote_identifier(self._cte_name(model_name))}.",
                )
            return rewritten

    def _build_measure_aggregation_sql(self, model_name: str, measure) -> str:
        """Build SQL aggregation expression for a measure.

        Note: Metric-level filters are already applied in the CTE via CASE WHEN
        on the raw column (see _build_model_cte). The raw column will have NULL
        for rows that don't match the filter. Therefore, we do NOT re-apply
        filters here - we just aggregate the pre-filtered raw column.

        This avoids the issue where re-applying filters on CTE columns could
        reference transformed values (e.g., DATE_TRUNC'd time dimensions)
        instead of the original raw values.

        Args:
            model_name: Name of the model containing the measure
            measure: Metric object representing the measure

        Returns:
            SQL aggregation expression string
        """
        agg_func = self._agg_sql_name(measure.agg)
        raw_col = self._cte_ref(model_name, f"{measure.name}_raw")

        # Simple aggregation - filters are already applied in CTE's raw column
        if agg_func == "COUNT_DISTINCT":
            return f"COUNT(DISTINCT {raw_col})"
        if agg_func == "COUNT":
            # Always use COUNT(raw_col) to avoid fan-out overcounting in multi-model joins.
            # The CTE projects a non-NULL raw column for matching rows, so COUNT(raw_col)
            # is equivalent to COUNT(*) for single-model queries but correct for joins.
            return f"COUNT({raw_col})"
        return f"{agg_func}({raw_col})"

    def _build_measure_window_total_sql(self, model_name: str, measure) -> str:
        """Build an all-rows window aggregate for a model-scoped measure."""
        agg_func = measure.agg.upper()
        raw_col = self._cte_ref(model_name, f"{measure.name}_raw")

        if agg_func == "COUNT_DISTINCT":
            return self._build_bsl_count_distinct_total_sql(model_name, measure)
        if agg_func == "COUNT":
            return f"SUM(COUNT({raw_col})) OVER ()"
        if agg_func == "SUM":
            return f"SUM(SUM({raw_col})) OVER ()"
        if agg_func == "AVG":
            return f"(SUM(SUM({raw_col})) OVER ()) / NULLIF(SUM(COUNT({raw_col})) OVER (), 0)"
        if agg_func == "MIN":
            return f"MIN(MIN({raw_col})) OVER ()"
        if agg_func == "MAX":
            return f"MAX(MAX({raw_col})) OVER ()"
        return f"SUM({self._build_measure_aggregation_sql(model_name, measure)}) OVER ()"

    def _build_measure_total_subquery_sql(self, model_name: str, measure) -> str:
        """Build a scalar all-rows aggregate from the model CTE."""
        cte_name = self._quote_identifier(self._cte_name(model_name))
        cte_alias = self._quote_identifier(f"{self._cte_name(model_name)}__all")
        raw_col = f"{cte_alias}.{self._quote_identifier(f'{measure.name}_raw')}"
        agg_func = measure.agg.upper()

        if agg_func == "COUNT_DISTINCT":
            expr = f"COUNT(DISTINCT {raw_col})"
        else:
            expr = self._build_measure_aggregation_sql(model_name, measure)
            expr = expr.replace(self._cte_ref(model_name, f"{measure.name}_raw"), raw_col)

        return f"(SELECT {expr} FROM {cte_name} AS {cte_alias})"

    def _build_bsl_count_distinct_total_sql(self, model_name: str, measure) -> str:
        """Build BSL _.all() SQL for a count-distinct measure."""
        context = getattr(self, "_bsl_all_query_context", None)
        if not context:
            return self._build_measure_total_subquery_sql(model_name, measure)

        base_model_name = context["base_model_name"]
        other_models = context["other_models"]
        models_with_filters = context["models_with_filters"]
        filters = context["filters"]
        query_model_names = [base_model_name] + other_models

        if model_name not in query_model_names:
            return self._build_measure_total_subquery_sql(model_name, measure)

        query = select(f"COUNT(DISTINCT {self._cte_ref(model_name, f'{measure.name}_raw')})").from_(
            self._quote_identifier(self._cte_name(base_model_name))
        )
        query = self._add_join_paths_to_query(query, base_model_name, other_models, models_with_filters)
        where_filters, _ = self._split_where_having_filters(filters, query_model_names)
        query = self._add_where_filters_to_query(query, where_filters, query_model_names)

        return f"({query.sql(dialect=self.dialect, pretty=True)})"

    def _extract_bsl_all_placeholders(self, formula: str) -> tuple[str, dict[str, str]]:
        """Replace BSL all-measure calls with placeholders before dependency replacement."""
        import re

        placeholders: dict[str, str] = {}

        def replace(match) -> str:
            placeholder = f"__bsl_all_{len(placeholders)}"
            placeholders[placeholder] = match.group(1).strip()
            return placeholder

        formula = re.sub(r"__bsl_all\(([A-Za-z_][A-Za-z0-9_.]*)\)", replace, formula)
        return formula, placeholders

    def _build_bsl_all_sql(self, metric_ref: str, model_context: str | None) -> str:
        """Resolve BSL _.all(metric) to a window total over the metric aggregate."""
        model_name = None
        measure_name = metric_ref

        if "." in metric_ref:
            model_name, measure_name = metric_ref.split(".", 1)
        elif model_context:
            model_name = model_context
        else:
            for candidate_name, candidate_model in self.graph.models.items():
                if candidate_model.get_metric(metric_ref):
                    model_name = candidate_name
                    break

        if not model_name:
            raise ValueError(f"Metric {metric_ref} not found")

        model = self.graph.get_model(model_name)
        measure = model.get_metric(measure_name)
        if not measure:
            raise ValueError(f"Metric {metric_ref} not found")
        if not measure.agg:
            return f"SUM({self._build_metric_sql(measure, model_name)}) OVER ()"
        return self._build_measure_window_total_sql(model_name, measure)

    def _build_metric_sql(self, metric, model_context: str | None = None) -> str:
        """Build SQL expression for a metric.

        Args:
            metric: Metric object
            model_context: Optional model name for resolving ambiguous references

        Returns:
            SQL expression string
        """
        if metric.type == "ratio":
            # numerator / NULLIF(denominator, 0)
            if not metric.numerator or not metric.denominator:
                raise ValueError(f"Ratio metric {metric.name} requires numerator and denominator")

            def resolve_ratio_ref(ref: str) -> str:
                if "." not in ref and model_context:
                    try:
                        context_model = self.graph.get_model(model_context)
                    except KeyError:
                        context_model = None

                    if context_model:
                        ref_metric = context_model.get_metric(ref)
                        if ref_metric:
                            if ref_metric.agg:
                                return self._build_measure_aggregation_sql(model_context, ref_metric)
                            return self._build_metric_sql(ref_metric, model_context)

                try:
                    ref_model, ref_metric = self.graph.resolve_metric_reference(ref)
                except KeyError as exc:
                    raise ValueError(f"Metric {ref} not found") from exc

                if ref_model:
                    if ref_metric.agg:
                        return self._build_measure_aggregation_sql(ref_model, ref_metric)
                    return self._build_metric_sql(ref_metric, ref_model)

                return self._build_metric_sql(ref_metric, model_context)

            num_expr = resolve_ratio_ref(metric.numerator)
            denom_expr = resolve_ratio_ref(metric.denominator)

            return f"({num_expr}) / NULLIF({denom_expr}, 0)"

        elif metric.agg:
            # Graph-level simple aggregations (e.g., SUM(orders.amount))
            # need CTE-qualified references.
            inner_expr = metric.sql or "*"
            if inner_expr != "*":
                inner_expr = self._rewrite_model_refs_to_ctes(inner_expr)

            if metric.agg == "count":
                if inner_expr == "*":
                    return "COUNT(*)"
                return f"COUNT({inner_expr})"
            if metric.agg == "count_distinct":
                return f"COUNT(DISTINCT {inner_expr})"
            return f"{self._agg_sql_name(metric.agg)}({inner_expr})"

        elif metric.type == "derived" or (not metric.type and not metric.agg and metric.sql):
            # Parse formula and replace metric references (handles both typed "derived" and untyped metrics with sql)
            if not metric.sql:
                raise ValueError(f"Derived metric {metric.name} missing sql")

            formula = metric.sql
            formula, bsl_all_placeholders = self._extract_bsl_all_placeholders(formula)

            # Check if this is a SQL expression metric (has inline aggregations)
            # These metrics already contain complete SQL and shouldn't have dependencies replaced
            has_inline_agg = sql_has_aggregate(formula, self.dialect)

            if has_inline_agg:
                # This is a SQL expression metric with inline aggregations.
                # Column references should already be qualified with {model} placeholder
                # at parse time (e.g., by SnowflakeAdapter).
                # We need to:
                # 1. Replace {model} with the CTE alias
                # 2. Replace measure column references with their _raw suffixed versions

                # Find which model this metric belongs to
                metric_model_name = model_context
                if not metric_model_name:
                    for m_name, m in self.graph.models.items():
                        if any(metric is mm for mm in m.metrics):
                            metric_model_name = m_name
                            break

                if metric_model_name:
                    cte_alias = self._quote_identifier(self._cte_name(metric_model_name))
                    # Replace {model} placeholder with the CTE alias
                    formula = formula.replace("{model}", cte_alias)

                    # Replace measure columns with their _raw suffixed versions
                    # For each measure in the model, check if it's referenced in the formula
                    model_obj = self.graph.get_model(metric_model_name)
                    if model_obj:
                        for measure in model_obj.metrics:
                            if measure.agg:  # Only process actual aggregation measures
                                # Replace cte_alias.measure_name with cte_alias.measure_name_raw
                                pattern = f"{cte_alias}.{measure.name}"
                                replacement = f"{cte_alias}.{measure.name}_raw"
                                formula = formula.replace(pattern, replacement)

                # Also rewrite any explicit model refs in the expression
                # (e.g., orders.amount -> orders_cte.amount) for graph metrics.
                formula = self._rewrite_model_refs_to_ctes(formula)
                return formula

            # Auto-detect dependencies from expression using graph for resolution
            dependencies = metric.get_dependencies(self.graph, model_context)

            # Sort dependencies by length descending to avoid partial matches
            # (e.g., replace "gross_revenue" before "revenue")
            sorted_deps = sorted(dependencies, key=len, reverse=True)

            # Replace each metric reference with its SQL expression
            for metric_name in sorted_deps:
                try:
                    ref_model_name, ref_metric = self.graph.resolve_metric_reference(metric_name)
                except KeyError:
                    ref_model_name = None
                    ref_metric = None

                if ref_metric and ref_model_name:
                    if ref_metric.agg:
                        # Use helper that applies metric-level filters
                        metric_sql = self._build_measure_aggregation_sql(ref_model_name, ref_metric)
                    else:
                        metric_sql = self._build_metric_sql(ref_metric, ref_model_name)
                elif ref_metric:
                    metric_sql = self._build_metric_sql(ref_metric, model_context)
                else:
                    # Try as graph-level metric
                    try:
                        ref_metric = self.graph.get_metric(metric_name)
                        # Recursively build metric SQL
                        metric_sql = self._build_metric_sql(ref_metric, model_context)
                    except KeyError:
                        raise ValueError(f"Metric {metric_name} not found")

                # Replace metric name in formula using word boundaries to avoid partial matches
                # Use regex to only replace whole word matches
                import re

                # For qualified names (model.measure), preserve explicit qualification.
                if "." in metric_name:
                    # Split into model and measure parts
                    parts = metric_name.split(".")
                    measure_only = parts[1]

                    # First try to replace qualified form if present.
                    pattern = r"\b" + re.escape(metric_name).replace(r"\.", r"\.") + r"\b"
                    if re.search(pattern, formula):
                        formula = re.sub(pattern, f"({metric_sql})", formula)
                    else:
                        # If the dependency resolved from an unqualified reference
                        # (e.g. count -> orders.count), replace only bare refs.
                        pattern = r"(?<!\.)\b" + re.escape(measure_only) + r"\b"
                        formula = re.sub(pattern, f"({metric_sql})", formula)
                else:
                    # For simple names, use word boundaries
                    pattern = r"(?<!\.)\b" + re.escape(metric_name) + r"\b"
                    formula = re.sub(pattern, f"({metric_sql})", formula)

            for placeholder, metric_ref in bsl_all_placeholders.items():
                formula = formula.replace(placeholder, f"({self._build_bsl_all_sql(metric_ref, model_context)})")

            return formula

        # Note: Cumulative metrics are handled via _generate_with_window_functions
        # and should never reach this point

        raise NotImplementedError(f"Metric type {metric.type} not yet implemented")

    def _generate_cohort_metric_query(
        self,
        metric_name: str,
        dimensions: list[str],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
    ) -> str:
        """Generate SQL for cohort metrics (two-level aggregation with HAVING).

        Pattern:
        1. Inner query: GROUP BY entity (+ entity_dimensions), compute inner_metrics, apply HAVING
        2. Outer query: re-aggregate filtered entities using the metric's agg

        Args:
            metric_name: Name of the cohort metric (can be "model.metric" format)
            dimensions: List of dimension references from the query
            filters: List of filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows
            offset: Number of rows to skip
        """
        import re as _re

        # Resolve the metric
        if "." in metric_name:
            model_name, local_name = metric_name.split(".", 1)
        else:
            local_name = metric_name
            model_name = None

        metric = None
        model = None
        if model_name:
            try:
                model = self.graph.get_model(model_name)
            except KeyError:
                pass
            if model:
                metric = model.get_metric(local_name)
        else:
            # Check graph-level metrics first
            try:
                metric = self.graph.get_metric(local_name)
            except KeyError:
                pass

        # Find the model that owns this metric if not already found
        if not model:
            if metric:
                # Check if any model owns this metric object
                for m_name, m in self.graph.models.items():
                    if any(metric is mm for mm in m.metrics):
                        model = m
                        model_name = m_name
                        break
            if not model:
                # Fall back to scanning models for a cohort metric by name
                matches = []
                for m_name, m in self.graph.models.items():
                    candidate = m.get_metric(local_name)
                    if candidate and candidate.type == "cohort":
                        matches.append((m_name, m, candidate))
                if len(matches) == 1:
                    model_name, model, metric = matches[0]
                elif len(matches) > 1:
                    raise ValueError(
                        f"Ambiguous cohort metric '{local_name}': found in models "
                        f"{', '.join(mn for mn, _, _ in matches)}. "
                        f"Use a model-qualified name (e.g., 'model_name.{local_name}') to disambiguate."
                    )

        if not model:
            # Last resort: find model by entity dimension match
            if metric and metric.entity:
                matching_models = []
                for m_name, m in self.graph.models.items():
                    for dim in m.dimensions:
                        if dim.name == metric.entity:
                            matching_models.append(m_name)
                            break
                if len(matching_models) == 1:
                    model = self.graph.get_model(matching_models[0])
                    model_name = matching_models[0]
                elif len(matching_models) > 1:
                    raise ValueError(
                        f"Ambiguous model for cohort metric '{metric_name}': "
                        f"entity dimension '{metric.entity}' found in multiple models: "
                        f"{', '.join(matching_models)}. "
                        f"Use a model-qualified metric name (e.g., 'model_name.{metric_name}') to disambiguate."
                    )

        if not model or not metric:
            raise ValueError(f"No model found for cohort metric {metric_name}")

        # Validate entity identifier
        if not _re.match(r"^[a-zA-Z_][a-zA-Z0-9_.]*$", metric.entity):
            raise ValueError(f"Invalid entity identifier: {metric.entity}")

        # Build FROM clause
        from_clause = self._model_from_clause(model)

        # Resolve entity SQL expression -- it might be a dimension name
        entity_sql = metric.entity
        entity_alias = metric.entity
        entity_dim = model.get_dimension(metric.entity)
        if entity_dim:
            entity_sql = entity_dim.sql_expr
            entity_alias = entity_dim.name

        def _replace_model_placeholder(expr: str) -> str:
            if model.sql:
                return expr.replace("{model}", "t")
            return expr.replace("{model}.", "")

        entity_sql = _replace_model_placeholder(entity_sql)

        # Process segments from filters
        all_filters = list(filters or [])
        segment_clause = ""
        remaining_filters = []
        for f in all_filters:
            # Check if filter references a segment
            segment_resolved = False
            for seg in model.segments or []:
                qualified = f"{model_name}.{seg.name}" if model_name else seg.name
                if f.strip() == qualified or f.strip() == seg.name:
                    seg_sql = _replace_model_placeholder(seg.sql)
                    segment_clause += f" AND ({seg_sql})"
                    segment_resolved = True
                    break
            if not segment_resolved:
                remaining_filters.append(f)

        # Build base WHERE from remaining filters
        filter_parts = []
        for f in remaining_filters:
            stripped = self._strip_model_prefixes([f], model_name or "")[0] if model_name else f
            stripped = _replace_model_placeholder(stripped)
            filter_parts.append(stripped)

        where_clause = "1=1" + segment_clause
        if filter_parts:
            where_clause += " AND " + " AND ".join(filter_parts)

        # Build inner GROUP BY columns
        quoted_entity = self._quote_alias(entity_alias)
        entity_select = f"{entity_sql} AS {quoted_entity}" if entity_sql != entity_alias else quoted_entity
        inner_group_cols = [entity_sql]
        inner_select_cols = [entity_select]

        # Entity dimensions (carried through to outer query)
        entity_dim_aliases = []
        for ed_name in metric.entity_dimensions or []:
            dim = model.get_dimension(ed_name)
            quoted_name = self._quote_alias(ed_name)
            if dim:
                dim_sql = _replace_model_placeholder(dim.sql_expr)
                inner_select_cols.append(f"{dim_sql} AS {quoted_name}")
                inner_group_cols.append(dim_sql)
                entity_dim_aliases.append(ed_name)
            else:
                inner_select_cols.append(quoted_name)
                inner_group_cols.append(quoted_name)
                entity_dim_aliases.append(ed_name)

        # Inner metrics
        inner_metric_selects = []
        for im in metric.inner_metrics:
            im_name = im["name"]
            im_agg = self._agg_sql_name(im.get("agg", "count"))
            im_sql = im.get("sql")

            if im_sql:
                im_sql = _replace_model_placeholder(im_sql)
                # Resolve dimension references in inner metric sql
                dim = model.get_dimension(im_sql)
                if dim:
                    im_sql = _replace_model_placeholder(dim.sql_expr)

            if not im_sql and im_agg != "COUNT":
                raise ValueError(
                    f"Inner metric '{im_name}' on cohort metric '{metric.name}' "
                    f"uses agg '{im_agg}' which requires a 'sql' field"
                )

            if im_agg == "COUNT_DISTINCT":
                expr = f"COUNT(DISTINCT {im_sql})"
            elif im_agg == "COUNT":
                expr = f"COUNT({im_sql or '*'})"
            else:
                expr = f"{im_agg}({im_sql})"

            inner_metric_selects.append(f"{expr} AS {self._quote_alias(im_name)}")

        having_clause = _replace_model_placeholder(metric.having)

        # Build outer query
        outer_agg = self._agg_sql_name(metric.agg)
        outer_sql = metric.sql
        if outer_sql:
            # Outer SQL references columns from the inner subquery (aliased as
            # cohort_sub), not the raw table alias used inside the subquery.
            outer_sql = outer_sql.replace("{model}.", "cohort_sub.").replace("{model}", "cohort_sub")

        if not outer_sql and outer_agg not in ("COUNT", "COUNT_DISTINCT"):
            raise ValueError(f"Cohort metric '{metric.name}' uses agg '{outer_agg}' which requires a 'sql' field")

        if outer_agg == "COUNT_DISTINCT":
            outer_expr = f"COUNT(DISTINCT {outer_sql or quoted_entity})"
        elif outer_agg == "COUNT":
            outer_expr = f"COUNT({outer_sql or '*'})"
        else:
            outer_expr = f"{outer_agg}({outer_sql})"

        outer_select_cols = []
        outer_group_cols = []

        # Add entity_dimensions to outer SELECT/GROUP BY
        for alias in entity_dim_aliases:
            quoted = self._quote_alias(alias)
            outer_select_cols.append(quoted)
            outer_group_cols.append(quoted)

        # Add query-level dimensions (if any beyond entity_dimensions)
        parsed_dims = self._parse_dimension_refs(dimensions)
        for dim_ref_str, granularity in parsed_dims:
            # Parse model.dimension or just dimension from the ref string
            if "." in dim_ref_str:
                dim_model_name, dim_name = dim_ref_str.split(".", 1)
            else:
                dim_model_name = model_name
                dim_name = dim_ref_str
            alias = dim_name

            if alias in entity_dim_aliases:
                continue  # Already included

            if dim_model_name != model_name:
                raise ValueError(
                    f"Cohort metric '{metric.name}' does not support dimensions "
                    f"from model '{dim_model_name}' (expected '{model_name}')"
                )
            dim = model.get_dimension(dim_name)
            if not dim:
                raise ValueError(f"Dimension '{dim_name}' not found on model '{model_name}'")
            dim_sql = _replace_model_placeholder(dim.sql_expr)
            if granularity:
                dim_sql = self._date_trunc(granularity, dim_sql)
            quoted_alias = self._quote_alias(alias)
            # Add to inner query so it's available in the outer subquery
            inner_select_cols.append(f"{dim_sql} AS {quoted_alias}")
            inner_group_cols.append(dim_sql)
            outer_select_cols.append(quoted_alias)
            outer_group_cols.append(quoted_alias)

        # Join inner select/group after dimensions are added
        inner_select = ",\n    ".join(inner_select_cols + inner_metric_selects)
        inner_group = ", ".join(inner_group_cols)

        outer_select_cols.append(f"{outer_expr} AS {self._quote_alias(metric.name)}")

        # Also add any additional outer metrics from inner_metrics that the user
        # might want (e.g., AVG(active_days) alongside COUNT)
        # For now, just output the primary metric

        outer_select = ",\n  ".join(outer_select_cols)

        group_by = ""
        if outer_group_cols:
            group_by = f"\nGROUP BY {', '.join(outer_group_cols)}"

        # Order/limit/offset
        order_clause = ""
        if order_by:
            order_fields = []
            for field in order_by:
                field_name = field.split(".", 1)[1] if "." in field else field
                # Handle "desc"/"asc" suffix
                parts = field_name.rsplit(" ", 1)
                if len(parts) == 2 and parts[1].upper() in ("ASC", "DESC"):
                    order_fields.append(f"{self._quote_alias(parts[0])} {parts[1].upper()}")
                else:
                    order_fields.append(self._quote_alias(field_name))
            order_clause = f"\nORDER BY {', '.join(order_fields)}"

        limit_clause = f"\nLIMIT {limit}" if limit is not None else ""
        offset_clause = f"\nOFFSET {offset}" if offset is not None else ""

        sql = f"""SELECT
  {outer_select}
FROM (
  SELECT
    {inner_select}
  FROM {from_clause}
  WHERE {where_clause}
  GROUP BY {inner_group}
  HAVING {having_clause}
) cohort_sub{group_by}{order_clause}{limit_clause}{offset_clause}"""

        return sql.strip()

    def _generate_retention_query(
        self,
        metric_name: str,
        dimensions: list[str],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
    ) -> str:
        """Generate SQL for cohort retention metrics.

        Uses a multi-CTE pattern:
        1. cohorts: identify each entity's first qualifying event (cohort_date)
        2. activity: distinct entity activity dates
        3. retention: join cohorts to activity, compute periods_since and active_users
        4. cohort_sizes: count of entities per cohort_date
        5. Final SELECT: retention percentage per cohort_date and period

        Args:
            metric_name: Name of the retention metric
            dimensions: List of dimension references (unused for retention, reserved)
            filters: List of filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows to return
            offset: Number of rows to skip (for pagination)

        Returns:
            SQL query string
        """
        import re as _re

        # Resolve metric and model
        metric = None
        model = None

        if "." in metric_name:
            model_name, measure_name = metric_name.split(".", 1)
            model = self.graph.get_model(model_name)
            if model:
                metric = model.get_metric(measure_name)
        else:
            try:
                metric = self.graph.get_metric(metric_name)
            except KeyError:
                pass

        if not metric or not metric.entity or not metric.cohort_event:
            raise ValueError(f"Retention metric {metric_name} missing required fields (entity, cohort_event)")

        # Find the model that owns this metric if not already found
        if not model:
            for m_name, m in self.graph.models.items():
                if any(metric is mm for mm in m.metrics):
                    model = m
                    break
            if not model:
                matching_models = []
                for m_name, m in self.graph.models.items():
                    for dim in m.dimensions:
                        if dim.name == metric.entity:
                            matching_models.append(m_name)
                            break
                if len(matching_models) == 1:
                    model = self.graph.get_model(matching_models[0])
                elif len(matching_models) > 1:
                    raise ValueError(
                        f"Ambiguous model for retention metric '{metric_name}': "
                        f"entity dimension '{metric.entity}' found in multiple models: "
                        f"{', '.join(matching_models)}. "
                        f"Use a model-qualified metric name (e.g., 'model_name.{metric_name}') to disambiguate."
                    )

        if not model:
            raise ValueError(f"No model found for retention metric {metric_name}")

        # Defaults (use `is not None` to avoid converting 0 to the default)
        periods = metric.periods if metric.periods is not None else 28
        granularity = metric.retention_granularity if metric.retention_granularity is not None else "day"
        activity_event = metric.activity_event or "TRUE"

        # Replace {model} placeholders in event predicates with actual table alias
        table_alias = "t" if model.sql else ""

        def _replace_model_placeholder(expr: str) -> str:
            if table_alias:
                return expr.replace("{model}", table_alias)
            else:
                return expr.replace("{model}.", "")

        cohort_event = _replace_model_placeholder(metric.cohort_event)
        activity_event = _replace_model_placeholder(activity_event)

        # Resolve entity dimension: use sql_expr for physical column, name for alias
        entity_dim = model.get_dimension(metric.entity)
        entity_sql = _replace_model_placeholder(entity_dim.sql_expr) if entity_dim else metric.entity
        entity_alias = metric.entity

        # Validate entity alias identifier
        if not _re.match(r"^[a-zA-Z_][a-zA-Z0-9_.]*$", entity_alias):
            raise ValueError(f"Invalid entity identifier: {entity_alias}")
        if not isinstance(periods, int) or periods < 1:
            raise ValueError(f"Invalid periods value: {periods}")

        # Find timestamp dimension: prefer model.default_time_dimension, fall back to first time dim
        timestamp_dim = None
        if model.default_time_dimension:
            timestamp_dim = model.get_dimension(model.default_time_dimension)
        if not timestamp_dim:
            for dim in model.dimensions:
                if dim.type == "time":
                    timestamp_dim = dim
                    break

        if not timestamp_dim:
            raise ValueError("Retention metrics require a time dimension on the model")

        # Use sql_expr for actual SQL, name for alias; expand {model} placeholder
        ts_sql = _replace_model_placeholder(timestamp_dim.sql_expr)

        # Build FROM clause
        from_clause = self._model_from_clause(model)

        # Build granularity-specific date truncation and date diff (dialect-aware)
        if granularity == "day":
            trunc_expr = f"CAST({ts_sql} AS DATE)"
            diff_expr = self._date_diff("a.active_date", "c.cohort_date", "day")
            periods_label = "days_since"
        elif granularity == "week":
            trunc_expr = f"CAST({self._date_trunc('week', ts_sql)} AS DATE)"
            diff_expr = f"{self._date_diff('a.active_date', 'c.cohort_date', 'day')} / 7"
            periods_label = "weeks_since"
        elif granularity == "month":
            trunc_expr = f"CAST({self._date_trunc('month', ts_sql)} AS DATE)"
            diff_expr = (
                "(EXTRACT(YEAR FROM a.active_date) - EXTRACT(YEAR FROM c.cohort_date)) * 12"
                " + (EXTRACT(MONTH FROM a.active_date) - EXTRACT(MONTH FROM c.cohort_date))"
            )
            periods_label = "months_since"
        else:
            raise ValueError(f"Unsupported retention granularity: {granularity}")

        # Merge metric-level filters with request-time filters
        all_filters = list(filters or [])
        if metric.filters:
            all_filters.extend(metric.filters)

        # Normalize filters: strip model name prefixes and resolve dimension names
        normalized_filters = self._strip_model_prefixes(all_filters, model.name)
        normalized_filters = self._resolve_filter_dimensions(normalized_filters, model)
        normalized_filters = [_replace_model_placeholder(f) for f in normalized_filters]

        # Build optional WHERE filters for the source data
        filter_clause = ""
        if normalized_filters:
            filter_clause = " AND " + " AND ".join(normalized_filters)

        order_clause = "\nORDER BY r.cohort_date, r.periods_since"
        if order_by:
            order_fields = []
            for field in order_by:
                field_name = field.split(".", 1)[1] if "." in field else field
                order_fields.append(field_name)
            order_clause = f"\nORDER BY {', '.join(order_fields)}"

        limit_clause = ""
        if limit is not None:
            limit_clause = f"\nLIMIT {limit}"

        offset_clause = ""
        if offset is not None:
            offset_clause = f"\nOFFSET {offset}"

        # Use entity_sql for raw-table references, entity_alias for CTE column names
        entity_select = f"{entity_sql} AS {entity_alias}" if entity_sql != entity_alias else entity_alias

        sql = f"""WITH cohorts AS (
  SELECT {entity_select}, MIN({trunc_expr}) AS cohort_date
  FROM {from_clause}
  WHERE {cohort_event}{filter_clause}
  GROUP BY {entity_sql}
),
activity AS (
  SELECT DISTINCT {entity_select}, {trunc_expr} AS active_date
  FROM {from_clause}
  WHERE {activity_event}{filter_clause}
),
retention AS (
  SELECT
    c.cohort_date,
    CAST({diff_expr} AS INTEGER) AS periods_since,
    COUNT(DISTINCT c.{entity_alias}) AS active_users
  FROM cohorts c
  JOIN activity a ON c.{entity_alias} = a.{entity_alias} AND a.active_date >= c.cohort_date
  WHERE CAST({diff_expr} AS INTEGER) <= {periods}
  GROUP BY 1, 2
),
cohort_sizes AS (
  SELECT cohort_date, COUNT(DISTINCT {entity_alias}) AS cohort_size
  FROM cohorts GROUP BY 1
)
SELECT
  r.cohort_date,
  r.periods_since AS {periods_label},
  r.active_users,
  c.cohort_size,
  ROUND(r.active_users * 100.0 / c.cohort_size, 1) AS retention_pct
FROM retention r
JOIN cohort_sizes c ON r.cohort_date = c.cohort_date{order_clause}{limit_clause}{offset_clause}"""

        return sql.strip()

    def _generate_conversion_query(
        self,
        metric_name: str,
        dimensions: list[str],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
    ) -> str:
        """Generate SQL for conversion funnel metrics.

        Supports two modes:
        - Legacy 2-step: uses base_event/conversion_event with self-join pattern
        - N-step funnel: uses steps list with per-entity BOOL_OR aggregation

        Args:
            metric_name: Name of the conversion metric (can be "metric" or "model.metric" format)
            dimensions: List of dimension references
            filters: List of filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows to return

        Returns:
            SQL query string
        """
        # Handle both "metric" and "model.metric" formats
        metric = None
        model = None

        if "." in metric_name:
            # model.metric format
            model_name, measure_name = metric_name.split(".", 1)
            model = self.graph.get_model(model_name)
            if model:
                metric = model.get_metric(measure_name)
        else:
            # Just metric name - try graph-level metric
            try:
                metric = self.graph.get_metric(metric_name)
            except KeyError:
                pass

        if not metric or not metric.entity:
            raise ValueError(f"Conversion metric {metric_name} missing required fields")

        # Determine if this is a multi-step funnel or legacy 2-step
        if metric.steps:
            return self._generate_multistep_conversion_query(metric, metric_name, dimensions, filters, order_by, limit)

        if not metric.base_event or not metric.conversion_event:
            raise ValueError(f"Conversion metric {metric_name} missing required fields")

        # Find the model that owns this metric if we haven't already
        if not model:
            # First, try to find which model has this conversion metric defined
            for m_name, m in self.graph.models.items():
                if any(metric is mm for mm in m.metrics):
                    model = m
                    break

            # If not found in any model, try to find a model with matching entity dimension
            if not model:
                for m_name, m in self.graph.models.items():
                    # Check if this model has a dimension matching the entity
                    for dim in m.dimensions:
                        if dim.name == metric.entity:
                            model = m
                            break
                    if model:
                        break

        if not model:
            raise ValueError(f"No model found for conversion metric {metric_name}")

        # Build SQL with self-join pattern
        # base_events: filter for base_event
        # conversion_events: filter for conversion_event
        # Join on entity where conversion is within window

        import re as _re

        window_parts = metric.conversion_window.split() if metric.conversion_window else ["7", "days"]
        window_num, window_unit = (
            window_parts[0],
            window_parts[1] if len(window_parts) > 1 else "days",
        )

        # Validate conversion metric fields to prevent SQL injection
        if not _re.match(r"^[a-zA-Z_][a-zA-Z0-9_.]*$", metric.entity):
            raise ValueError(f"Invalid entity identifier: {metric.entity}")
        if not _re.match(r"^\d+$", window_num):
            raise ValueError(f"Invalid window number: {window_num}")
        if not _re.match(r"^[a-zA-Z]+$", window_unit):
            raise ValueError(f"Invalid window unit: {window_unit}")

        # Find dimension that represents event type
        event_type_dim = None
        timestamp_dim = None
        for dim in model.dimensions:
            if dim.type == "time":
                timestamp_dim = dim.name
            # Assume event_type is categorical
            if "event" in dim.name.lower() and "type" in dim.name.lower():
                event_type_dim = dim.name

        if not event_type_dim or not timestamp_dim:
            raise ValueError("Conversion metrics require event_type and timestamp dimensions")

        # Build FROM clause - handle both SQL and table-backed models
        from_clause = self._model_from_clause(model)

        # Normalize filters: strip model name prefixes and resolve dimension names
        normalized_filters = self._strip_model_prefixes(filters or [], model.name)
        normalized_filters = self._resolve_filter_dimensions(normalized_filters, model)
        filter_clause = ""
        if normalized_filters:
            filter_clause = "\n    AND " + " AND ".join(f"({f})" for f in normalized_filters)

        # Resolve dimension columns for GROUP BY support.
        # Each entry is (alias, sql_expr) where sql_expr uses the dimension's
        # actual SQL expression and applies DATE_TRUNC for time granularity.
        dim_entries: list[tuple[str, str]] = []
        for dim_ref in dimensions:
            dim_name = dim_ref.split(".", 1)[1] if "." in dim_ref else dim_ref
            # Parse granularity suffix (e.g., "event_date__month")
            if "__" in dim_name:
                base_dim, gran = dim_name.rsplit("__", 1)
            else:
                base_dim, gran = dim_name, None
            dim_obj = model.get_dimension(base_dim)
            if not dim_obj:
                continue
            sql_col = dim_obj.sql_expr
            if gran and dim_obj.type == "time":
                sql_col = self._date_trunc(gran, sql_col)
                alias = f"{base_dim}__{gran}"
            else:
                alias = base_dim
            dim_entries.append((alias, sql_col))

        dim_aliases = [alias for alias, _ in dim_entries]

        # Build extra SELECT columns for dimensions
        extra_base_cols = ""
        extra_conv_cols = ""
        extra_conversions_cols = ""
        if dim_entries:
            base_col_list = ",\n    ".join(f"{sql_col} AS {alias}" for alias, sql_col in dim_entries)
            extra_base_cols = f",\n    {base_col_list}"
            extra_conv_cols = extra_base_cols
            conv_col_list = ",\n    ".join(f"base.{alias}" for alias in dim_aliases)
            extra_conversions_cols = f",\n    {conv_col_list}"

        # Build LEFT JOIN condition (entity + dimensions)
        join_on_parts = ["base_events.entity = conversions.entity"]
        for alias in dim_aliases:
            join_on_parts.append(f"base_events.{alias} IS NOT DISTINCT FROM conversions.{alias}")
        join_condition = "\n  AND ".join(join_on_parts)

        # Build final SELECT, GROUP BY, ORDER BY, LIMIT
        dim_select = ""
        group_by = ""
        if dim_aliases:
            dim_select_list = ",\n  ".join(f"base_events.{alias} AS {alias}" for alias in dim_aliases)
            dim_select = f"  {dim_select_list},\n"
            group_by = "\nGROUP BY\n  " + ",\n  ".join(str(i + 1) for i in range(len(dim_aliases)))

        order_clause = ""
        if order_by:
            order_fields = []
            for field in order_by:
                field_name = field.split(".", 1)[1] if "." in field else field
                order_fields.append(field_name)
            order_clause = f"\nORDER BY {', '.join(order_fields)}"

        limit_clause = ""
        if limit is not None:
            limit_clause = f"\nLIMIT {limit}"

        sql = f"""
WITH base_events AS (
  SELECT
    {metric.entity} AS entity,
    {timestamp_dim} AS event_time{extra_base_cols}
  FROM {from_clause}
  WHERE {event_type_dim} = '{metric.base_event.replace("'", "''")}'{filter_clause}
),
conversion_events AS (
  SELECT
    {metric.entity} AS entity,
    {timestamp_dim} AS event_time{extra_conv_cols}
  FROM {from_clause}
  WHERE {event_type_dim} = '{metric.conversion_event.replace("'", "''")}'{filter_clause}
),
conversions AS (
  SELECT DISTINCT
    base.entity{extra_conversions_cols}
  FROM base_events base
  JOIN conversion_events conv
    ON base.entity = conv.entity
    AND conv.event_time BETWEEN base.event_time AND base.event_time + {self._build_interval(window_num, window_unit)}
)
SELECT
{dim_select}  COUNT(DISTINCT conversions.entity)::FLOAT / NULLIF(COUNT(DISTINCT base_events.entity), 0) AS {metric.name}
FROM base_events
LEFT JOIN conversions ON {join_condition}{group_by}{order_clause}{limit_clause}
"""

        return sql.strip()

    def _generate_multistep_conversion_query(
        self,
        metric,
        metric_name: str,
        dimensions: list[str],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
    ) -> str:
        """Generate SQL for N-step conversion funnel metrics.

        Uses a sequential CTE chain: each step CTE joins back to the previous
        step and only considers events AFTER the prior step's timestamp. This
        correctly handles entities with repeated actions (e.g., an entity that
        does step 2 before step 1 is only counted if they also do step 2 after
        step 1).

        Args:
            metric: The Metric object with steps defined
            metric_name: Name of the conversion metric
            dimensions: List of dimension references
            filters: List of filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows to return

        Returns:
            SQL query string
        """
        import re as _re

        # Validate entity identifier
        if not _re.match(r"^[a-zA-Z_][a-zA-Z0-9_.]*$", metric.entity):
            raise ValueError(f"Invalid entity identifier: {metric.entity}")

        # Find the model that owns this metric
        model = None
        if "." in metric_name:
            model_name, _ = metric_name.split(".", 1)
            model = self.graph.get_model(model_name)
        if not model:
            for m_name, m in self.graph.models.items():
                if m.get_metric(metric_name.split(".", 1)[-1] if "." in metric_name else metric_name):
                    model = m
                    break
            if not model:
                for m_name, m in self.graph.models.items():
                    for dim in m.dimensions:
                        if dim.name == metric.entity:
                            model = m
                            break
                    if model:
                        break

        if not model:
            raise ValueError(f"No model found for conversion metric {metric_name}")

        # Find timestamp dimension: prefer model.default_time_dimension, fall back to first time dim
        timestamp_dim = None
        if model.default_time_dimension:
            timestamp_dim = model.get_dimension(model.default_time_dimension)
        if not timestamp_dim:
            for dim in model.dimensions:
                if dim.type == "time":
                    timestamp_dim = dim
                    break

        if not timestamp_dim:
            raise ValueError(
                f"Multi-step conversion funnel requires a time dimension on model '{model.name}' "
                "to enforce chronological step ordering"
            )

        ts_sql_raw = timestamp_dim.sql_expr

        # Build FROM clause
        from_clause = self._model_from_clause(model)

        # Normalize SQL expressions for use inside subqueries.
        # Expressions may contain {model} placeholders (e.g. "{model}.created_at")
        # or model-name prefixes (e.g. "events.created_at") that are invalid
        # inside the step subqueries which use different aliases.
        def _normalize_expr_for_subquery(sql_expr: str, table_alias: str, qualify_bare: bool = False) -> str:
            """Rewrite {model} placeholders and model-name prefixes for a subquery.

            For SQL-backed models the table_alias (e.g. "t" or "_src") replaces
            the placeholder.  For table-backed models the prefix is stripped.

            When qualify_bare=True and table_alias is set, unqualified column
            references are prefixed with the alias (e.g. ``event`` -> ``s.event``).
            This prevents ambiguous-column errors in JOINed CTEs.
            """
            result = sql_expr
            if table_alias:
                result = result.replace("{model}", table_alias)
            else:
                result = result.replace("{model}.", "")
            # Rewrite column references via sqlglot to avoid corrupting string
            # literals (e.g. "events.signup" inside a quoted value).
            try:
                parsed = sqlglot.parse_one(result, dialect=self.dialect)
                for col in parsed.find_all(exp.Column):
                    if col.table == model.name:
                        # Strip model-name qualifier (e.g. events.col -> col)
                        # or replace with alias when qualify_bare is set
                        col.set("table", table_alias if qualify_bare and table_alias else None)
                    elif not col.table and qualify_bare and table_alias:
                        col.set("table", table_alias)
                result = parsed.sql(dialect=self.dialect)
            except Exception:
                pass
            return result

        # ts_sql for step 1 (from_clause aliases SQL models as "t", table models have no alias)
        ts_sql = _normalize_expr_for_subquery(ts_sql_raw, "t" if model.sql else "")

        # ts_sql for from_clause_s subquery (SQL models aliased as "_src", table models no alias)
        ts_sql_src = _normalize_expr_for_subquery(ts_sql_raw, "_src" if model.sql else "")

        # Resolve entity: use dimension sql_expr if available (handles aliased dims
        # and qualified names like events.user_id -> user_id)
        entity_dim = model.get_dimension(metric.entity)
        entity_sql_raw = entity_dim.sql_expr if entity_dim else metric.entity
        # Normalize for step 1 context (alias "t" for SQL models, bare for table models)
        entity_sql = _normalize_expr_for_subquery(entity_sql_raw, "t" if model.sql else "")
        # Normalize for step N context (always alias "s")
        entity_sql_s = _normalize_expr_for_subquery(entity_sql_raw, "s", qualify_bare=True)

        # Normalize filters: strip model name prefixes and resolve dimension names
        normalized_filters = self._strip_model_prefixes(filters or [], model.name)
        normalized_filters = self._resolve_filter_dimensions(normalized_filters, model)

        # Build WHERE filter clauses for step 1 and step N
        # Step N needs bare columns qualified with "s." to avoid ambiguity in JOINs
        filter_clause = ""
        filter_clause_s = ""
        if normalized_filters:
            filter_clause = " AND " + " AND ".join(f"({f})" for f in normalized_filters)
            qualified = [_normalize_expr_for_subquery(f, "s", qualify_bare=True) for f in normalized_filters]
            filter_clause_s = " AND " + " AND ".join(f"({q})" for q in qualified)

        # Resolve dimension columns for GROUP BY support
        dim_entries: list[tuple[str, str]] = []
        for dim_ref in dimensions:
            dim_name = dim_ref.split(".", 1)[1] if "." in dim_ref else dim_ref
            if "__" in dim_name:
                base_dim, gran = dim_name.rsplit("__", 1)
            else:
                base_dim, gran = dim_name, None
            dim_obj = model.get_dimension(base_dim)
            if not dim_obj:
                continue
            sql_col = _normalize_expr_for_subquery(dim_obj.sql_expr, "t" if model.sql else "")
            if gran and dim_obj.type == "time":
                sql_col = self._date_trunc(gran, sql_col)
                alias = f"{base_dim}__{gran}"
            else:
                alias = base_dim
            dim_entries.append((alias, sql_col))

        dim_aliases = [alias for alias, _ in dim_entries]

        # Build from_clause variant for non-first steps (aliased as 's' for joining).
        # Project the timestamp expression as __ts so later CTEs can reference
        # s.__ts regardless of whether ts_sql is a simple column or an expression
        # like CAST(created_at AS DATE).
        if model.sql:
            from_clause_s = f"(SELECT *, {ts_sql_src} AS __ts FROM ({model.sql}) AS _src) AS s"
        else:
            from_clause_s = f"(SELECT *, {ts_sql_src} AS __ts FROM {model.table}) AS s"

        # Build sequential CTE chain: each step derives its timestamp from the
        # prior step's completion, ensuring correct chronological ordering.
        cte_parts = []
        num_steps = len(metric.steps)

        for i, step_expr in enumerate(metric.steps, 1):
            if i == 1:
                # Normalize step predicate: rewrite {model} / model-name prefixes
                # for step 1 scope (SQL models alias as "t", table models have no alias)
                norm_step = _normalize_expr_for_subquery(step_expr, "t" if model.sql else "")

                # Step 1: find the earliest matching event per entity
                select_parts = [f"{entity_sql} AS entity", f"MIN({ts_sql}) AS step_1_ts"]
                for alias, sql_col in dim_entries:
                    select_parts.append(f"{sql_col} AS {alias}")
                select_str = ",\n    ".join(select_parts)
                group_parts = [entity_sql]
                for _alias, sql_col in dim_entries:
                    group_parts.append(sql_col)
                group_str = ",\n    ".join(group_parts)
                cte_parts.append(
                    f"step_1 AS (\n"
                    f"  SELECT\n"
                    f"    {select_str}\n"
                    f"  FROM {from_clause}\n"
                    f"  WHERE ({norm_step}){filter_clause}\n"
                    f"  GROUP BY\n"
                    f"    {group_str}\n"
                    f")"
                )
            else:
                # Normalize step predicate: rewrite {model} / model-name prefixes
                # for step N scope. Source is always aliased as "s" (both SQL and
                # table-backed models), so qualify_bare with "s" unconditionally.
                norm_step = _normalize_expr_for_subquery(step_expr, "s", qualify_bare=True)

                # Step N: join source to step N-1, only consider events at or after prior step
                prev = f"step_{i - 1}"
                select_parts = [f"{entity_sql_s} AS entity", f"MIN(s.__ts) AS step_{i}_ts"]
                for alias, _sql_col in dim_entries:
                    select_parts.append(f"{prev}.{alias}")
                select_str = ",\n    ".join(select_parts)
                group_parts = [entity_sql_s]
                for alias, _sql_col in dim_entries:
                    group_parts.append(f"{prev}.{alias}")
                group_str = ",\n    ".join(group_parts)
                cte_parts.append(
                    f"step_{i} AS (\n"
                    f"  SELECT\n"
                    f"    {select_str}\n"
                    f"  FROM {from_clause_s}\n"
                    f"  JOIN {prev} ON {entity_sql_s} = {prev}.entity\n"
                    f"    AND s.__ts >= {prev}.step_{i - 1}_ts\n"
                    f"  WHERE ({norm_step}){filter_clause_s}\n"
                    f"  GROUP BY\n"
                    f"    {group_str}\n"
                    f")"
                )

        # Build final SELECT: count distinct entities from each step CTE
        # Also alias the last step count to the metric name for ORDER BY compatibility
        metric_name_only = metric_name.split(".", 1)[-1] if "." in metric_name else metric_name
        final_select_parts = []
        for alias in dim_aliases:
            final_select_parts.append(f"step_1.{alias}")
        final_select_parts.append("COUNT(DISTINCT step_1.entity) AS total_entities")
        for i in range(1, num_steps + 1):
            final_select_parts.append(f"COUNT(DISTINCT step_{i}.entity) AS step_{i}_count")
        # Add metric-named column (last step count) so ORDER BY metric_name works
        final_select_parts.append(f"COUNT(DISTINCT step_{num_steps}.entity) AS {metric_name_only}")
        final_select = ",\n  ".join(final_select_parts)

        # Build LEFT JOIN chain: step_1 LEFT JOIN step_2 LEFT JOIN step_3 ...
        join_parts = []
        for i in range(2, num_steps + 1):
            join_on = f"step_{i - 1}.entity = step_{i}.entity"
            for alias in dim_aliases:
                join_on += f"\n    AND step_{i - 1}.{alias} IS NOT DISTINCT FROM step_{i}.{alias}"
            join_parts.append(f"LEFT JOIN step_{i} ON {join_on}")
        join_str = "\n".join(join_parts)

        # Build GROUP BY, ORDER BY, LIMIT for final query
        final_group_by = ""
        if dim_aliases:
            final_group_by = "\nGROUP BY\n  " + ",\n  ".join(str(i + 1) for i in range(len(dim_aliases)))

        order_clause = ""
        if order_by:
            order_fields = []
            for field in order_by:
                field_name = field.split(".", 1)[1] if "." in field else field
                order_fields.append(field_name)
            order_clause = f"\nORDER BY {', '.join(order_fields)}"

        limit_clause = ""
        if limit is not None:
            limit_clause = f"\nLIMIT {limit}"

        # Assemble full query
        cte_str = ",\n".join(cte_parts)
        join_section = f"\n{join_str}" if join_parts else ""
        sql = f"""
WITH {cte_str}
SELECT
  {final_select}
FROM step_1{join_section}{final_group_by}{order_clause}{limit_clause}
"""

        return sql.strip()

    def _generate_with_window_functions(
        self,
        metrics: list[str],
        dimensions: list[str],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
        aliases: dict[str, str] | None = None,
        use_preaggregations: bool = False,
    ) -> str:
        """Generate SQL with window functions for cumulative metrics.

        This uses a subquery pattern:
        1. Inner query aggregates base metrics by dimensions
        2. Outer query applies window functions to aggregated results

        Args:
            metrics: List of metric references
            dimensions: List of dimension references
            filters: List of filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows to return
            aliases: Custom aliases for fields (dict mapping field reference to alias)
            use_preaggregations: Try materialized pre-aggregation routing for the inner base query

        Returns:
            SQL query string
        """
        aliases = aliases or {}
        # Separate window function metrics from regular metrics
        cumulative_metrics = []
        time_comparison_metrics = []
        offset_ratio_metrics = []
        conversion_metrics = []
        retention_metrics = []
        cohort_metrics = []
        base_metrics = []
        time_comparison_base_plans = {}
        regular_expression_metric_plans = {}
        precomputed_cumulative_metrics = []

        def add_unique(items: list[str], value: str | None) -> None:
            if value and value not in items:
                items.append(value)

        def metric_ref_alias(metric_ref: str) -> str:
            return metric_ref.split(".", 1)[1] if "." in metric_ref else metric_ref

        def canonical_ref(metric_ref: str, model_context: str | None = None) -> str:
            if "." in metric_ref:
                return metric_ref

            if model_context:
                try:
                    model = self.graph.get_model(model_context)
                except KeyError:
                    model = None
                if model and model.get_metric(metric_ref):
                    return f"{model_context}.{metric_ref}"

            return metric_ref

        def resolve_metric_ref(metric_ref: str, model_context: str | None = None):
            if "." in metric_ref:
                model_name, metric_name = metric_ref.split(".", 1)
                try:
                    model = self.graph.get_model(model_name)
                except KeyError:
                    model = None
                if model:
                    model_metric = model.get_metric(metric_name)
                    if model_metric:
                        return model_metric, model_name
                try:
                    return self.graph.get_metric(metric_ref), model_context
                except KeyError:
                    return None, model_context

            if model_context:
                try:
                    context_model = self.graph.get_model(model_context)
                except KeyError:
                    context_model = None
                if context_model:
                    context_metric = context_model.get_metric(metric_ref)
                    if context_metric:
                        return context_metric, model_context

            try:
                return self.graph.get_metric(metric_ref), model_context
            except KeyError:
                pass

            # Fall back to scanning models for the metric by name
            matches = []
            for m_name, m in self.graph.models.items():
                found = m.get_metric(metric_ref)
                if found:
                    matches.append((found, m_name))
            if len(matches) == 1:
                return matches[0]
            if len(matches) > 1:
                model_names = ", ".join(mn for _, mn in matches)
                raise ValueError(
                    f"Ambiguous metric '{metric_ref}': found in models {model_names}. "
                    f"Use a model-qualified name (e.g., 'model_name.{metric_ref}') to disambiguate."
                )

            return None, model_context

        def collect_leaf_base_metrics(
            metric_ref: str, model_context: str | None = None, visited: set[str] | None = None
        ) -> None:
            if visited is None:
                visited = set()

            canon_ref = canonical_ref(metric_ref, model_context)
            if canon_ref in visited:
                return
            visited.add(canon_ref)

            metric_obj, resolved_context = resolve_metric_ref(canon_ref, model_context)
            if not metric_obj:
                add_unique(base_metrics, canon_ref)
                return

            is_expression_metric = not metric_obj.type and not metric_obj.agg and metric_obj.sql
            if metric_obj.type == "derived" or is_expression_metric:
                dependencies = metric_obj.get_dependencies(self.graph, resolved_context)
                if not dependencies:
                    add_unique(base_metrics, canon_ref)
                    return
                for dep in dependencies:
                    collect_leaf_base_metrics(dep, resolved_context, visited)
                return

            if metric_obj.type == "ratio" and not metric_obj.offset_window:
                if metric_obj.numerator:
                    collect_leaf_base_metrics(metric_obj.numerator, resolved_context, visited)
                if metric_obj.denominator:
                    collect_leaf_base_metrics(metric_obj.denominator, resolved_context, visited)
                return

            add_unique(base_metrics, canon_ref)
            if metric_obj.type == "cumulative":
                add_unique(precomputed_cumulative_metrics, canon_ref)

        for m in metrics:
            metric_context = m.split(".", 1)[0] if "." in m else None
            metric, resolved_context = resolve_metric_ref(m, metric_context)

            # Classify metric by type
            if metric and metric.type == "cumulative":
                add_unique(cumulative_metrics, m)
                # Add the base measure/metric to base_metrics
                if metric.sql:
                    add_unique(base_metrics, canonical_ref(metric.sql, resolved_context))
                elif metric.base_metric:
                    add_unique(base_metrics, canonical_ref(metric.base_metric, resolved_context))
            elif metric and metric.type == "time_comparison":
                # Validate required fields
                if not metric.base_metric:
                    raise ValueError(f"time_comparison metric '{m}' requires 'base_metric' field")
                add_unique(time_comparison_metrics, m)

                base_ref = canonical_ref(metric.base_metric, resolved_context)
                base_metric, base_context = resolve_metric_ref(base_ref, resolved_context)
                base_is_expression_metric = bool(
                    base_metric
                    and (
                        base_metric.type == "derived"
                        or (not base_metric.type and not base_metric.agg and base_metric.sql)
                        or (base_metric.type == "ratio" and not base_metric.offset_window)
                    )
                )

                if base_is_expression_metric:
                    collect_leaf_base_metrics(base_ref, base_context)
                else:
                    add_unique(base_metrics, base_ref)

                time_comparison_base_plans[m] = {
                    "base_ref": base_ref,
                    "base_context": base_context,
                    "is_expression": base_is_expression_metric,
                }
            elif metric and metric.type == "ratio" and metric.offset_window:
                add_unique(offset_ratio_metrics, m)
                # Add numerator and denominator to base_metrics
                if metric.numerator:
                    add_unique(base_metrics, canonical_ref(metric.numerator, resolved_context))
                if metric.denominator:
                    add_unique(base_metrics, canonical_ref(metric.denominator, resolved_context))
            elif metric and metric.type == "conversion":
                add_unique(conversion_metrics, m)
                # Conversion metrics need special handling - don't add to base_metrics
            elif metric and metric.type == "retention":
                add_unique(retention_metrics, m)
                # Retention metrics need special handling - don't add to base_metrics
            elif metric and metric.type == "cohort":
                add_unique(cohort_metrics, m)
                # Cohort metrics need special handling - don't add to base_metrics
            else:
                # Regular metric or measure
                metric_ref = canonical_ref(m, resolved_context)
                is_expression_metric = bool(
                    metric
                    and (
                        metric.type == "derived"
                        or (metric.type == "ratio" and not metric.offset_window)
                        or (not metric.type and not metric.agg and metric.sql)
                    )
                )
                if is_expression_metric:
                    collect_leaf_base_metrics(metric_ref, resolved_context)
                    regular_expression_metric_plans[m] = {
                        "metric_ref": metric_ref,
                        "metric_context": resolved_context,
                    }
                else:
                    add_unique(base_metrics, metric_ref)

        # When cumulative dependencies are needed to build time comparison base
        # expressions, precompute them in the inner query and skip re-computing
        # them in the outer window pass.
        if precomputed_cumulative_metrics:
            cumulative_metrics = [m for m in cumulative_metrics if m not in precomputed_cumulative_metrics]

        # Handle retention metrics separately - they need a completely different pattern
        if retention_metrics:
            if len(retention_metrics) > 1:
                raise ValueError("Only one retention metric can be queried at a time")
            non_retention = (
                base_metrics
                or cumulative_metrics
                or time_comparison_metrics
                or offset_ratio_metrics
                or conversion_metrics
                or cohort_metrics
                or regular_expression_metric_plans
            )
            if non_retention:
                raise ValueError("Retention metrics cannot be combined with other metrics in a single query")
            return self._generate_retention_query(retention_metrics[0], dimensions, filters, order_by, limit, offset)

        # Handle conversion metrics separately - they need a completely different pattern
        if conversion_metrics:
            non_conversion = (
                base_metrics
                or cumulative_metrics
                or time_comparison_metrics
                or offset_ratio_metrics
                or cohort_metrics
                or retention_metrics
                or regular_expression_metric_plans
            )
            if non_conversion:
                raise ValueError("Conversion metrics cannot be combined with other metrics in a single query")
            return self._generate_conversion_query(conversion_metrics[0], dimensions, filters, order_by, limit)

        # Handle cohort metrics separately - two-level aggregation pattern
        if cohort_metrics:
            if len(cohort_metrics) > 1:
                raise ValueError("Only one cohort metric can be queried at a time")
            non_cohort = (
                base_metrics
                or cumulative_metrics
                or time_comparison_metrics
                or offset_ratio_metrics
                or conversion_metrics
                or retention_metrics
                or regular_expression_metric_plans
            )
            if non_cohort:
                raise ValueError("Cohort metrics cannot be combined with other metrics in a single query")
            return self._generate_cohort_metric_query(cohort_metrics[0], dimensions, filters, order_by, limit, offset)

        # Build inner query with base aggregations
        # Dedupe base_metrics to avoid duplicate column names
        base_metrics = list(dict.fromkeys(base_metrics))
        base_metric_aliases = {metric_ref_alias(m) for m in base_metrics}

        inner_query = self.generate(
            metrics=base_metrics,
            dimensions=dimensions,
            filters=filters,
            order_by=None,  # Apply ordering in outer query
            limit=None,  # Apply limit in outer query
            use_preaggregations=use_preaggregations,
        )

        # Parse dimensions for outer SELECT
        parsed_dims = self._parse_dimension_refs(dimensions)

        def sql_identifier(name: str) -> str:
            import re

            if re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name):
                return name
            return self._quote_alias(name)

        def metric_column(metric_ref: str) -> str:
            alias = metric_ref_alias(metric_ref)
            return f"base.{sql_identifier(alias)}"

        def build_time_comparison_base_expression(
            metric_ref: str, model_context: str | None = None, visited: set[str] | None = None
        ) -> str:
            import re

            if visited is None:
                visited = set()

            canon_ref = canonical_ref(metric_ref, model_context)
            if canon_ref in visited:
                return metric_column(canon_ref)
            visited.add(canon_ref)

            metric_obj, resolved_context = resolve_metric_ref(canon_ref, model_context)
            if not metric_obj:
                return metric_column(canon_ref)

            is_expression_metric = not metric_obj.type and not metric_obj.agg and metric_obj.sql
            if metric_obj.type == "derived" or is_expression_metric:
                if not metric_obj.sql:
                    return metric_column(canon_ref)

                formula = metric_obj.sql
                dependencies = sorted(metric_obj.get_dependencies(self.graph, resolved_context), key=len, reverse=True)
                if not dependencies:
                    # Dependency-free expression metrics are already materialized
                    # in the inner query (when needed), so reuse the alias.
                    return metric_column(canon_ref)
                for dependency in dependencies:
                    dep_expr = build_time_comparison_base_expression(dependency, resolved_context, visited.copy())
                    if "." in dependency:
                        model_name, dep_name = dependency.split(".", 1)
                        qualified_pattern = r"\b" + re.escape(f"{model_name}.{dep_name}") + r"\b"
                        if re.search(qualified_pattern, formula):
                            formula = re.sub(qualified_pattern, f"({dep_expr})", formula)
                        else:
                            unqualified_pattern = r"\b" + re.escape(dep_name) + r"\b"
                            formula = re.sub(unqualified_pattern, f"({dep_expr})", formula)
                    else:
                        dep_pattern = r"\b" + re.escape(dependency) + r"\b"
                        formula = re.sub(dep_pattern, f"({dep_expr})", formula)
                return formula

            if metric_obj.type == "ratio" and not metric_obj.offset_window:
                if not metric_obj.numerator or not metric_obj.denominator:
                    raise ValueError(f"Ratio metric {metric_obj.name} requires numerator and denominator")
                num_expr = build_time_comparison_base_expression(metric_obj.numerator, resolved_context, visited.copy())
                denom_expr = build_time_comparison_base_expression(
                    metric_obj.denominator, resolved_context, visited.copy()
                )
                return f"({num_expr}) / NULLIF({denom_expr}, 0)"

            return metric_column(canon_ref)

        # Build outer SELECT with window functions
        select_exprs = []

        # Add dimensions
        for dim_ref, gran in parsed_dims:
            # Inner query uses simple alias without model prefix
            dim_name = dim_ref.split(".")[1] if "." in dim_ref else dim_ref
            alias = dim_name
            if gran:
                alias = f"{alias}__{gran}"
            select_exprs.append(f"base.{alias}")

        # Add base metrics (pass through)
        for m in base_metrics:
            alias = metric_ref_alias(m)
            select_exprs.append(f"base.{sql_identifier(alias)}")

        # Add expression metrics that depend on cumulative/derived chains.
        for plan in regular_expression_metric_plans.values():
            metric_ref = plan["metric_ref"]
            if metric_ref_alias(metric_ref) in base_metric_aliases:
                continue
            metric_expr = build_time_comparison_base_expression(metric_ref, plan["metric_context"])
            select_exprs.append(f"{metric_expr} AS {sql_identifier(metric_ref_alias(metric_ref))}")

        # Track cumulative window expressions for inclusion in LAG CTE path
        cumulative_window_entries = []  # (window_expr, alias)

        # Add cumulative metrics with window functions
        for m in cumulative_metrics:
            # Handle both qualified (model.measure) and unqualified references
            metric = None
            if "." in m:
                model_name, measure_name = m.split(".", 1)
                try:
                    model = self.graph.get_model(model_name)
                    metric = model.get_metric(measure_name) if model else None
                except KeyError:
                    pass
                # Fall back to graph-level metric with dotted name
                if not metric:
                    try:
                        metric = self.graph.get_metric(m)
                    except KeyError:
                        pass
                # Use just the measure name as alias if it's model.measure, otherwise full name
                # Quote to handle any special characters
                metric_alias = self._quote_alias(measure_name if metric and "." not in metric.name else m)
            else:
                metric = self.graph.get_metric(m)
                # Quote to handle dotted metric names
                metric_alias = self._quote_alias(m)
            if not metric or (not metric.sql and not metric.window_expression):
                continue

            # Find the time dimension to order by
            time_dim = None
            for dim_ref, gran in parsed_dims:
                dim_name = dim_ref.split(".")[1] if "." in dim_ref else dim_ref
                model_name = dim_ref.split(".")[0] if "." in dim_ref else None
                if model_name:
                    model = self.graph.get_model(model_name)
                    if model:
                        dim = model.get_dimension(dim_name)
                        if dim and dim.type == "time":
                            # Use simple alias without model prefix
                            time_dim = f"base.{dim_name}"
                            if gran:
                                time_dim = f"base.{dim_name}__{gran}"
                            break

            # Allow window_order to override auto-detected time dimension
            if metric.window_order:
                time_dim = f"base.{metric.window_order}"

            if not time_dim:
                raise ValueError(f"Cumulative metric {m} requires a time dimension for ordering")

            # Option C: Raw window_expression passthrough
            if metric.window_expression:
                order_col = time_dim
                frame = metric.window_frame or "ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"
                window_expr = f"{metric.window_expression} OVER (ORDER BY {order_col} {frame}) AS {metric_alias}"
                select_exprs.append(window_expr)
                cumulative_window_entries.append((window_expr, metric_alias))
                continue

            # Option A: Use agg + sql (supports AVG, COUNT, etc.)
            # Get base measure/metric to apply window function to
            base_ref = metric.sql
            if "." in base_ref:
                # It's a direct measure reference - extract just the measure name
                base_alias = base_ref.split(".")[1]
            else:
                # It's an unqualified reference - check model first, then graph-level
                base_metric = None
                # Get model name from the cumulative metric reference
                cum_model_name = m.split(".")[0] if "." in m else None
                if cum_model_name:
                    cum_model = self.graph.get_model(cum_model_name)
                    if cum_model:
                        base_metric = cum_model.get_metric(base_ref)

                # Fallback to graph-level metric
                if not base_metric:
                    try:
                        base_metric = self.graph.get_metric(base_ref)
                    except KeyError:
                        pass

                if base_metric and base_metric.sql:
                    # Window over the base query's metric alias, not the raw column behind that metric.
                    base_alias = base_metric.name
                else:
                    # Fallback to the metric name itself
                    base_alias = base_ref

            # Determine aggregation function (default to SUM for backwards compatibility)
            agg_func = self._agg_sql_name(metric.agg or "sum")
            if agg_func == "COUNT_DISTINCT":
                agg_func = "COUNT"
                base_col = f"DISTINCT base.{base_alias}"
            else:
                base_col = f"base.{base_alias}"

            # Build window function
            if metric.grain_to_date:
                # Grain-to-date: MTD, QTD, YTD
                # Partition by the grain period and order within it
                grain = metric.grain_to_date
                partition = self._date_trunc(grain, time_dim)

                window_expr = f"{agg_func}({base_col}) OVER (PARTITION BY {partition} ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS {metric_alias}"
            elif metric.window:
                # Parse window (e.g., "7 days")
                window_parts = metric.window.split()
                if len(window_parts) == 2:
                    num, unit = window_parts
                    # For date-based windows, use RANGE
                    window_expr = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} RANGE BETWEEN INTERVAL '{num} {unit}' PRECEDING AND CURRENT ROW) AS {metric_alias}"
                else:
                    # Fallback to rows
                    window_expr = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS {metric_alias}"
            else:
                # Running total (unbounded window)
                window_expr = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS {metric_alias}"

            select_exprs.append(window_expr)
            cumulative_window_entries.append((window_expr, metric_alias))

        # Add time comparison metrics with LAG window functions
        # Note: We'll handle these with a CTE approach similar to offset_ratio_metrics
        # to avoid nested window functions

        # Handle offset ratio metrics OR time comparison metrics - need two-step CTEs
        if offset_ratio_metrics or time_comparison_metrics:
            # Need an intermediate CTE with LAG values
            lag_selects = []

            # Track what columns we're selecting
            lag_cte_columns = []

            # Add dimensions and base metrics
            for dim_ref, gran in parsed_dims:
                dim_name = dim_ref.split(".")[1] if "." in dim_ref else dim_ref
                alias = dim_name
                if gran:
                    alias = f"{alias}__{gran}"
                lag_selects.append(f"base.{sql_identifier(alias)}")
                lag_cte_columns.append(alias)

            for m in base_metrics:
                alias = metric_ref_alias(m)
                lag_selects.append(f"base.{sql_identifier(alias)}")
                lag_cte_columns.append(alias)

            for plan in regular_expression_metric_plans.values():
                metric_ref = plan["metric_ref"]
                metric_expr = build_time_comparison_base_expression(metric_ref, plan["metric_context"])
                metric_expr_alias = metric_ref_alias(metric_ref)
                if metric_expr_alias in lag_cte_columns:
                    continue
                lag_selects.append(f"{metric_expr} AS {sql_identifier(metric_expr_alias)}")
                lag_cte_columns.append(metric_expr_alias)

            # Include cumulative window expressions in the LAG CTE
            for cum_expr, cum_alias in cumulative_window_entries:
                lag_selects.append(cum_expr)
                lag_cte_columns.append(cum_alias)

            # Add LAG expressions for each time comparison metric
            for m in time_comparison_metrics:
                metric_context = m.split(".", 1)[0] if "." in m else None
                metric, _ = resolve_metric_ref(m, metric_context)
                if not metric or not metric.base_metric:
                    continue

                # Find time dimension
                time_dim = None
                time_dim_gran = None
                for dim_ref, gran in parsed_dims:
                    dim_name = dim_ref.split(".")[1] if "." in dim_ref else dim_ref
                    model_name = dim_ref.split(".")[0] if "." in dim_ref else None
                    if model_name:
                        model = self.graph.get_model(model_name)
                        if model:
                            dim = model.get_dimension(dim_name)
                            if dim and dim.type == "time":
                                time_dim = f"base.{dim_name}"
                                if gran:
                                    time_dim = f"base.{dim_name}__{gran}"
                                    time_dim_gran = gran
                                break

                if not time_dim:
                    raise ValueError(f"Time comparison metric {m} requires a time dimension")

                # Partition by non-time dimensions to avoid cross-group leakage.
                partition_cols = []
                for partition_dim_ref, partition_gran in parsed_dims:
                    partition_dim_name = (
                        partition_dim_ref.split(".")[1] if "." in partition_dim_ref else partition_dim_ref
                    )
                    partition_alias = partition_dim_name
                    if partition_gran:
                        partition_alias = f"{partition_alias}__{partition_gran}"
                    partition_col = f"base.{partition_alias}"
                    if partition_col != time_dim and partition_col not in partition_cols:
                        partition_cols.append(partition_col)
                partition_clause = ""
                if partition_cols:
                    partition_clause = f"PARTITION BY {', '.join(partition_cols)} "

                base_plan = time_comparison_base_plans.get(m, {})
                base_ref = base_plan.get("base_ref", metric.base_metric)
                base_context = base_plan.get("base_context")
                base_alias = metric_ref_alias(base_ref)
                base_is_expression = bool(base_plan.get("is_expression"))

                if base_is_expression and base_alias not in lag_cte_columns:
                    base_expr = build_time_comparison_base_expression(base_ref, base_context)
                    lag_selects.append(f"{base_expr} AS {sql_identifier(base_alias)}")
                    lag_cte_columns.append(base_alias)

                if base_is_expression and base_alias in base_metric_aliases:
                    lag_input_expr = metric_column(base_ref)
                elif base_is_expression:
                    lag_input_expr = build_time_comparison_base_expression(base_ref, base_context)
                else:
                    lag_input_expr = metric_column(base_ref)

                # Calculate LAG offset
                lag_offset = self._calculate_lag_offset(metric.comparison_type, time_dim_gran)

                # Add LAG for base metric (quote alias to handle dotted names)
                prev_value_alias = self._quote_alias(f"{m}_prev_value")
                lag_selects.append(
                    f"LAG({lag_input_expr}, {lag_offset}) OVER ({partition_clause}ORDER BY {time_dim}) AS {prev_value_alias}"
                )

            # Add LAG expressions for each offset ratio metric
            for m in offset_ratio_metrics:
                metric_context = m.split(".", 1)[0] if "." in m else None
                metric, _ = resolve_metric_ref(m, metric_context)
                if not metric or not metric.numerator or not metric.denominator:
                    continue

                # Find time dimension
                time_dim = None
                for dim_ref, gran in parsed_dims:
                    dim_name = dim_ref.split(".")[1] if "." in dim_ref else dim_ref
                    model_name = dim_ref.split(".")[0] if "." in dim_ref else None
                    if model_name:
                        model = self.graph.get_model(model_name)
                        if model:
                            dim = model.get_dimension(dim_name)
                            if dim and dim.type == "time":
                                time_dim = f"base.{dim_name}"
                                if gran:
                                    time_dim = f"base.{dim_name}__{gran}"
                                break

                if not time_dim:
                    raise ValueError(f"Offset ratio metric {m} requires a time dimension")

                # Partition by non-time dimensions to avoid cross-group leakage.
                partition_cols = []
                for partition_dim_ref, partition_gran in parsed_dims:
                    partition_dim_name = (
                        partition_dim_ref.split(".")[1] if "." in partition_dim_ref else partition_dim_ref
                    )
                    partition_alias = partition_dim_name
                    if partition_gran:
                        partition_alias = f"{partition_alias}__{partition_gran}"
                    partition_col = f"base.{partition_alias}"
                    if partition_col != time_dim and partition_col not in partition_cols:
                        partition_cols.append(partition_col)
                partition_clause = ""
                if partition_cols:
                    partition_clause = f"PARTITION BY {', '.join(partition_cols)} "

                # Get denominator alias
                denom_alias = metric_ref_alias(metric.denominator)

                # Add LAG for denominator - reference base.denom_alias since it's from inner query
                # Quote alias to handle dotted names
                prev_denom_alias = self._quote_alias(f"{m}_prev_denom")
                lag_selects.append(
                    f"LAG(base.{sql_identifier(denom_alias)}) OVER ({partition_clause}ORDER BY {time_dim}) AS {prev_denom_alias}"
                )

            # Build intermediate CTE - inner_query already has all the columns we need
            # We need to add "base." prefix since we're wrapping inner_query in a FROM (inner_query) AS base
            lag_selects_str = ",\n    ".join(lag_selects)
            lag_cte_sql = f"WITH lag_cte AS (\n  SELECT\n    {lag_selects_str}\n  FROM (\n{inner_query}\n  ) AS base\n)"

            # Now build final select from lag_cte - need to rebuild select_exprs without base. prefix
            final_selects = []

            # Re-add dimensions and base metrics from lag_cte
            for col in lag_cte_columns:
                final_selects.append(col)

            # Add time comparison metrics
            for m in time_comparison_metrics:
                metric_context = m.split(".", 1)[0] if "." in m else None
                metric, _ = resolve_metric_ref(m, metric_context)
                if not metric or not metric.base_metric:
                    continue

                base_plan = time_comparison_base_plans.get(m, {})
                base_ref = base_plan.get("base_ref", metric.base_metric)
                base_alias = sql_identifier(metric_ref_alias(base_ref))

                # Quote aliases to handle dotted metric names
                prev_value_col = self._quote_alias(f"{m}_prev_value")
                final_alias = self._quote_alias(m)

                # Build calculation based on calculation type
                calc_type = metric.calculation or "percent_change"
                if calc_type == "difference":
                    expr = f"({base_alias} - {prev_value_col}) AS {final_alias}"
                elif calc_type == "percent_change":
                    expr = f"(({base_alias} - {prev_value_col}) / NULLIF({prev_value_col}, 0) * 100) AS {final_alias}"
                elif calc_type == "ratio":
                    expr = f"({base_alias} / NULLIF({prev_value_col}, 0)) AS {final_alias}"
                else:
                    raise ValueError(f"Unknown calculation type: {calc_type}")

                final_selects.append(expr)

            # Add offset ratio metrics
            for m in offset_ratio_metrics:
                metric_context = m.split(".", 1)[0] if "." in m else None
                metric, _ = resolve_metric_ref(m, metric_context)
                if not metric:
                    continue

                num_alias = sql_identifier(metric_ref_alias(metric.numerator))

                # Quote aliases to handle dotted metric names
                prev_denom_col = self._quote_alias(f"{m}_prev_denom")
                final_alias = self._quote_alias(m)

                # Calculate ratio using the lagged value
                offset_expr = f"{num_alias} / NULLIF({prev_denom_col}, 0) AS {final_alias}"
                final_selects.append(offset_expr)

            # Build final query
            final_select_str = ",\n  ".join(final_selects)
            outer_query = f"{lag_cte_sql}\nSELECT\n  {final_select_str}\nFROM lag_cte"
        else:
            # Build outer query without LAG CTE
            select_expr_str = ",\n  ".join(select_exprs)
            outer_query = f"SELECT\n  {select_expr_str}\nFROM (\n{inner_query}\n) AS base"

        # Add ORDER BY if specified
        if order_by:
            order_clauses = []
            for field in order_by:
                if "." in field:
                    # Extract just the field name without model prefix
                    field_alias = field.split(".")[1]
                else:
                    field_alias = field
                order_clauses.append(field_alias)
            outer_query += f"\nORDER BY {', '.join(order_clauses)}"

        # Add LIMIT and OFFSET if specified
        if limit is not None:
            outer_query += f"\nLIMIT {limit}"
        if offset is not None:
            outer_query += f"\nOFFSET {offset}"

        return outer_query

    def _split_preaggregation_filters(self, model, filters: list[str]) -> tuple[list[str], list[str]]:
        row_filters: list[str] = []
        aggregate_filters: list[str] = []
        for filter_expr in filters:
            if self._filter_references_model_metric(model, filter_expr):
                aggregate_filters.append(filter_expr)
            else:
                row_filters.append(filter_expr)
        return row_filters, aggregate_filters

    def _filter_references_model_metric(self, model, filter_expr: str) -> bool:
        try:
            parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
        except Exception:
            return False

        for column in parsed.find_all(exp.Column):
            table_name = column.table.replace("_cte", "") if column.table else None
            field_name = column.name.rsplit("__", 1)[0] if "__" in column.name else column.name
            if table_name and table_name != model.name:
                continue
            if model.get_metric(field_name):
                return True
        return False

    def _preaggregation_metric_expression(self, model, preagg, metric_name: str) -> str:
        metric = model.get_metric(metric_name)
        raw_col = f"{metric_name}_raw"
        if not metric:
            return raw_col
        if metric.type == "ratio":
            numerator = self._preaggregation_metric_expression(model, preagg, (metric.numerator or "").split(".")[-1])
            denominator = self._preaggregation_metric_expression(
                model,
                preagg,
                (metric.denominator or "").split(".")[-1],
            )
            return f"{numerator} / NULLIF({denominator}, 0)"
        if metric.type == "derived" or (not metric.type and not metric.agg and metric.sql):
            return self._preaggregation_derived_metric_expression(model, preagg, metric)
        if metric.agg in {"sum", "count"}:
            return f"SUM({raw_col})"
        if metric.agg == "avg":
            matcher = PreAggregationMatcher(model)
            count_measure = matcher._find_count_measure_for_avg(metric, preagg.measures or []) or "count"
            return f"SUM({raw_col}) / NULLIF(SUM({count_measure}_raw), 0)"
        if metric.agg in {"min", "max"}:
            return f"{metric.agg.upper()}({raw_col})"
        return f"SUM({raw_col})"

    def _preaggregation_derived_metric_expression(self, model, preagg, metric) -> str:
        if not metric.sql:
            return f"SUM({metric.name}_raw)"
        try:
            expression = sqlglot.parse_one(metric.sql, dialect=self.dialect)
        except Exception:
            return metric.sql

        def replace_column(node):
            if not isinstance(node, exp.Column):
                return node
            field_name = node.name
            if model.get_metric(field_name):
                return sqlglot.parse_one(
                    self._preaggregation_metric_expression(model, preagg, field_name),
                    dialect=self.dialect,
                )
            return node

        return expression.transform(replace_column).sql(dialect=self.dialect)

    def _rewrite_preaggregation_aggregate_filter(self, model, preagg, filter_expr: str) -> str:
        try:
            parsed = sqlglot.parse_one(filter_expr, dialect=self.dialect)
        except Exception:
            return filter_expr.replace(f"{model.name}_cte.", "").replace(f"{model.name}.", "")

        def replace_column(node):
            if not isinstance(node, exp.Column):
                return node

            table_name = node.table.replace("_cte", "") if node.table else None
            if table_name and table_name != model.name:
                return node

            field_name = node.name.rsplit("__", 1)[0] if "__" in node.name else node.name
            if model.get_metric(field_name):
                return sqlglot.parse_one(
                    self._preaggregation_metric_expression(model, preagg, field_name),
                    dialect=self.dialect,
                )

            if preagg.time_dimension and preagg.granularity and field_name == preagg.time_dimension:
                return exp.column(f"{preagg.time_dimension}_{preagg.granularity}")
            return exp.column(node.name)

        return parsed.transform(replace_column).sql(dialect=self.dialect)

    def explain_join_key_preaggregation(
        self,
        metrics: list[str],
        dimensions: list[str],
        filters: list[str] | None = None,
        aliases: dict[str, str] | None = None,
    ) -> dict[str, object]:
        route, reason = self._join_key_preaggregation_route(metrics, dimensions, filters or [], aliases or {})
        if route is None:
            return {"eligible": False, "reason": reason}

        preagg = route["preagg"]
        join_path = route["join_path"]
        return {
            "eligible": True,
            "reason": "matching_join_key_preaggregation",
            "model": route["metric_model_name"],
            "remote_model": route["remote_model_name"],
            "preaggregation": preagg.name,
            "preaggregation_dimensions": route["preaggregation_dimensions"],
            "join_keys": [
                {
                    "from_model": hop.from_model,
                    "to_model": hop.to_model,
                    "from_columns": hop.from_columns,
                    "to_columns": hop.to_columns,
                    "relationship": hop.relationship,
                }
                for hop in join_path
            ],
        }

    def _try_use_join_key_preaggregation(
        self,
        metrics: list[str],
        dimensions: list[str],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
        aliases: dict[str, str] | None = None,
    ) -> str | None:
        route, _reason = self._join_key_preaggregation_route(metrics, dimensions, filters or [], aliases or {})
        if route is None:
            return None
        return self._generate_join_key_preaggregation(
            route=route,
            metrics=metrics,
            dimensions=dimensions,
            order_by=order_by,
            limit=limit,
            offset=offset,
            aliases=aliases or {},
        )

    def _join_key_preaggregation_route(
        self,
        metrics: list[str],
        dimensions: list[str],
        filters: list[str],
        aliases: dict[str, str],
    ) -> tuple[dict[str, object] | None, str]:
        if filters:
            return None, "filters_not_supported"

        metric_models = []
        metric_names = []
        for metric_ref in metrics:
            if "." not in metric_ref:
                return None, "graph_metrics_not_supported"
            model_name, metric_name = metric_ref.split(".", 1)
            if model_name not in metric_models:
                metric_models.append(model_name)
            metric_names.append(metric_name)

        if len(metric_models) != 1:
            return None, "not_single_metric_model_query"
        if not dimensions:
            return None, "no_remote_dimensions"

        metric_model_name = metric_models[0]
        metric_model = self.graph.get_model(metric_model_name)
        if not metric_model.pre_aggregations:
            return None, "model_has_no_preaggregations"

        parsed_dims = self._parse_dimension_refs(dimensions)
        local_preagg_dimensions: list[str] = []
        local_time_granularity: str | None = None
        remote_model_name: str | None = None
        join_path = None
        remote_dimensions: list[tuple[str, str | None]] = []
        join_keys: list[str] = []

        for dim_ref, granularity in parsed_dims:
            if "." not in dim_ref:
                return None, "unqualified_dimensions_not_supported"
            dim_model_name, dim_name = dim_ref.split(".", 1)
            if dim_model_name == metric_model_name:
                local_preagg_dimensions.append(dim_name)
                dimension = metric_model.get_dimension(dim_name)
                if dimension and dimension.type == "time":
                    effective_granularity = granularity or dimension.granularity
                    if local_time_granularity and local_time_granularity != effective_granularity:
                        return None, "multiple_local_time_granularities_not_supported"
                    local_time_granularity = effective_granularity
                continue

            try:
                path = self.graph.find_relationship_path(metric_model_name, dim_model_name)
            except (KeyError, ValueError):
                return None, "remote_dimension_not_reachable"

            if len(path) != 1:
                return None, "multi_hop_remote_dimension_not_supported"
            if path[0].relationship != "many_to_one":
                return None, "remote_dimension_not_many_to_one"
            if remote_model_name and remote_model_name != dim_model_name:
                return None, "multiple_remote_models_not_supported"

            remote_model_name = dim_model_name
            join_path = path
            remote_dimensions.append((dim_ref, granularity))
            for join_key in path[0].from_columns:
                if join_key not in join_keys:
                    join_keys.append(join_key)

        if not remote_model_name or not join_path:
            return None, "no_remote_dimensions"

        preaggregation_dimensions = [*local_preagg_dimensions, *join_keys]
        matcher = PreAggregationMatcher(metric_model)
        preagg = matcher.find_matching_preagg(
            metrics=metric_names,
            dimensions=preaggregation_dimensions,
            time_granularity=local_time_granularity,
            filters=[],
        )
        if not preagg:
            return None, "missing_join_key_preaggregation"

        return (
            {
                "metric_model_name": metric_model_name,
                "metric_model": metric_model,
                "remote_model_name": remote_model_name,
                "remote_model": self.graph.get_model(remote_model_name),
                "join_path": join_path,
                "preagg": preagg,
                "parsed_dims": parsed_dims,
                "remote_dimensions": remote_dimensions,
                "preaggregation_dimensions": preaggregation_dimensions,
                "aliases": aliases,
            },
            "matching_join_key_preaggregation",
        )

    def _generate_join_key_preaggregation(
        self,
        route: dict[str, object],
        metrics: list[str],
        dimensions: list[str],
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
        aliases: dict[str, str] | None = None,
    ) -> str:
        aliases = aliases or {}
        metric_model = route["metric_model"]
        metric_model_name = route["metric_model_name"]
        remote_model = route["remote_model"]
        remote_model_name = route["remote_model_name"]
        preagg = route["preagg"]
        join_path = route["join_path"]
        parsed_dims = route["parsed_dims"]

        rollup_alias = f"{metric_model_name}_rollup"
        remote_alias = remote_model_name
        preagg_table = preagg.get_table_name(
            metric_model.name,
            database=self.preagg_database,
            schema=self.preagg_schema,
        )

        select_exprs: list[str] = []
        output_names: dict[str, str] = {}

        def register_output_name(output_name: str, *refs: str) -> None:
            for ref in refs:
                if ref:
                    output_names[ref] = output_name

        def dimension_output_name(dim_ref: str, dim_name: str, granularity: str | None) -> str:
            full_ref = f"{dim_ref}__{granularity}" if granularity else dim_ref
            default = f"{dim_name}__{granularity}" if granularity else dim_name
            return aliases.get(full_ref) or aliases.get(dim_ref) or default

        for dim_ref, granularity in parsed_dims:
            dim_model_name, dim_name = dim_ref.split(".", 1)
            output_name = dimension_output_name(dim_ref, dim_name, granularity)
            full_ref = f"{dim_ref}__{granularity}" if granularity else dim_ref
            default_name = f"{dim_name}__{granularity}" if granularity else dim_name
            register_output_name(output_name, full_ref, dim_ref, dim_name, default_name, output_name)

            if dim_model_name == metric_model_name:
                column_name = dim_name
                if preagg.time_dimension == dim_name and preagg.granularity:
                    column_name = f"{dim_name}_{preagg.granularity}"
                column_expr = f"{rollup_alias}.{self._quote_identifier(column_name)}"
            else:
                dimension = remote_model.get_dimension(dim_name)
                dimension_sql = dimension.sql if dimension and dimension.sql else dim_name
                column_expr = self._qualify_model_expression(dimension_sql, remote_model_name, remote_alias)

            if granularity:
                column_expr = self._date_trunc(granularity, column_expr)
            select_exprs.append(f"{column_expr} AS {self._quote_alias(output_name)}")

        for metric_ref in metrics:
            _model_name, metric_name = metric_ref.split(".", 1)
            output_name = aliases.get(metric_ref) or metric_name
            register_output_name(output_name, metric_ref, metric_name, output_name)
            metric_expr = self._preaggregation_metric_expression(metric_model, preagg, metric_name)
            metric_expr = self._qualify_model_expression(metric_expr, metric_model_name, rollup_alias)
            select_exprs.append(f"{metric_expr} AS {self._quote_alias(output_name)}")

        hop = join_path[0]
        join_conditions = [
            f"{rollup_alias}.{self._quote_identifier(from_column)} = {remote_alias}.{self._quote_identifier(to_column)}"
            for from_column, to_column in zip(hop.from_columns, hop.to_columns)
        ]

        group_by_clause = ""
        if parsed_dims:
            group_by_clause = "\nGROUP BY " + ", ".join(str(i) for i in range(1, len(parsed_dims) + 1))

        order_by_clause = ""
        if order_by:
            order_clauses = []
            for field in order_by:
                parts = field.rsplit(" ", 1)
                direction = ""
                field_ref = field
                if len(parts) == 2 and parts[1].upper() in {"ASC", "DESC"}:
                    field_ref = parts[0]
                    direction = f" {parts[1].upper()}"
                field_name = field_ref.split(".", 1)[1] if "." in field_ref else field_ref
                output_name = output_names.get(field_ref) or output_names.get(field_name) or field_ref
                order_clauses.append(f"{self._quote_alias(output_name)}{direction}")
            order_by_clause = "\nORDER BY " + ", ".join(order_clauses)

        limit_clause = ""
        if limit is not None:
            limit_clause = f"\nLIMIT {limit}"
        if offset is not None:
            limit_clause += f"\nOFFSET {offset}"

        select_exprs_str = ",\n  ".join(select_exprs)
        remote_source = self._model_source_as(remote_model, remote_alias)
        return f"""SELECT
  {select_exprs_str}
FROM {remote_source}
LEFT JOIN {preagg_table} AS {rollup_alias}
  ON {" AND ".join(join_conditions)}{group_by_clause}{order_by_clause}{limit_clause}"""

    def _qualify_model_expression(self, expression_sql: str, model_name: str, alias: str) -> str:
        expression_sql = expression_sql.replace("{model}", alias)
        try:
            expression = sqlglot.parse_one(expression_sql, dialect=self.dialect)
        except Exception:
            if self._is_simple_identifier(expression_sql):
                return f"{alias}.{self._quote_identifier(expression_sql)}"
            return expression_sql

        for column in expression.find_all(exp.Column):
            if not column.table:
                column.set("table", exp.to_identifier(alias))
            elif column.table == model_name:
                column.set("table", exp.to_identifier(alias))
        return expression.sql(dialect=self.dialect)

    def _try_use_preaggregation(
        self,
        model_name: str,
        metrics: list[str],
        parsed_dims: list[tuple[str, str | None]],
        filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
        aliases: dict[str, str] | None = None,
    ) -> str | None:
        """Try to generate query using a pre-aggregation.

        Args:
            model_name: Name of the model to query
            metrics: List of metric references
            parsed_dims: Parsed dimension references with granularities
            filters: List of filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows
            offset: Number of rows to skip

        Returns:
            SQL query string if pre-aggregation found, None otherwise
        """
        model = self.graph.get_model(model_name)
        if not model or not model.pre_aggregations:
            return None

        # Extract dimension names (without model prefix) and find time granularity
        dim_names = []
        time_granularity = None

        for dim_ref, gran in parsed_dims:
            # Remove model prefix if present
            if "." in dim_ref:
                dim_name = dim_ref.split(".", 1)[1]
            else:
                dim_name = dim_ref

            dim_names.append(dim_name)

            # Track time granularity for matching
            if gran:
                time_granularity = gran

        # Extract metric names (without model prefix)
        metric_names = []
        for m in metrics:
            if "." in m:
                metric_name = m.split(".", 1)[1]
            else:
                metric_name = m
            metric_names.append(metric_name)

        row_filters, aggregate_filters = self._split_preaggregation_filters(model, filters or [])

        # Try to find matching pre-aggregation. Only row-stage filters constrain
        # matching dimensions; aggregate-stage metric filters are rendered as
        # HAVING over derivable rollup measures.
        filter_exprs = []
        if row_filters:
            for f in row_filters:
                # Strip model prefix for filter compatibility check
                # e.g., "orders.status = 'completed'" -> "status = 'completed'"
                filter_expr = f.replace(f"{model_name}.", "").replace(f"{model_name}_cte.", "")
                filter_exprs.append(filter_expr)

        matcher = PreAggregationMatcher(model)
        preagg = matcher.find_matching_preagg(
            metrics=metric_names,
            dimensions=dim_names,
            time_granularity=time_granularity,
            filters=filter_exprs,
        )

        if not preagg:
            return None

        # Generate SQL against pre-aggregation table
        return self._generate_from_preaggregation(
            model=model,
            preagg=preagg,
            metrics=metrics,
            parsed_dims=parsed_dims,
            filters=row_filters,
            aggregate_filters=aggregate_filters,
            order_by=order_by,
            limit=limit,
            offset=offset,
            aliases=aliases,
        )

    def _generate_from_preaggregation(
        self,
        model,
        preagg,
        metrics: list[str],
        parsed_dims: list[tuple[str, str | None]],
        filters: list[str] | None = None,
        aggregate_filters: list[str] | None = None,
        order_by: list[str] | None = None,
        limit: int | None = None,
        offset: int | None = None,
        aliases: dict[str, str] | None = None,
    ) -> str:
        """Generate SQL query from a pre-aggregation.

        Args:
            model: The model containing the pre-aggregation
            preagg: The pre-aggregation to use
            metrics: List of metric references
            parsed_dims: Parsed dimension references with granularities
            filters: List of row-stage filter expressions
            aggregate_filters: List of metric-stage filter expressions
            order_by: List of fields to order by
            limit: Maximum number of rows
            offset: Number of rows to skip

        Returns:
            SQL query string
        """
        aliases = aliases or {}
        preagg_table = preagg.get_table_name(model.name, database=self.preagg_database, schema=self.preagg_schema)

        # Build SELECT clause
        select_exprs = []
        output_names: dict[str, str] = {}

        def register_output_name(output_name: str, *refs: str) -> None:
            for ref in refs:
                if ref:
                    output_names[ref] = output_name

        def dimension_output_name(dim_ref: str, dim_name: str, gran: str | None) -> str:
            if gran:
                full_ref = f"{dim_ref}__{gran}"
                default = f"{dim_name}__{gran}"
            else:
                full_ref = dim_ref
                default = dim_name

            canonical_ref = full_ref if "." in full_ref else f"{model.name}.{full_ref}"
            return aliases.get(full_ref) or aliases.get(canonical_ref) or default

        def metric_output_name(metric_ref: str, metric_name: str) -> str:
            canonical_ref = metric_ref if "." in metric_ref else f"{model.name}.{metric_ref}"
            return aliases.get(metric_ref) or aliases.get(canonical_ref) or metric_name

        # Add dimensions
        for dim_ref, gran in parsed_dims:
            # Get dimension name without model prefix
            if "." in dim_ref:
                dim_name = dim_ref.split(".", 1)[1]
            else:
                dim_name = dim_ref

            output_name = dimension_output_name(dim_ref, dim_name, gran)
            full_ref = f"{dim_ref}__{gran}" if gran else dim_ref
            canonical_ref = full_ref if "." in full_ref else f"{model.name}.{full_ref}"
            default_name = f"{dim_name}__{gran}" if gran else dim_name
            register_output_name(output_name, full_ref, canonical_ref, dim_name, default_name, output_name)

            # Check if this is the time dimension and needs granularity conversion
            if gran and preagg.time_dimension == dim_name:
                # Need to convert from pre-agg granularity to query granularity
                preagg_col = f"{dim_name}_{preagg.granularity}"

                if gran == preagg.granularity:
                    # Exact match - use as is with proper alias
                    select_exprs.append(f"{preagg_col} AS {self._quote_alias(output_name)}")
                else:
                    # Roll up to coarser granularity
                    date_trunc_expr = self._date_trunc(gran, preagg_col)
                    select_exprs.append(f"{date_trunc_expr} AS {self._quote_alias(output_name)}")
            else:
                # Regular dimension - use as is
                if output_name == dim_name:
                    select_exprs.append(f"{dim_name}")
                else:
                    select_exprs.append(f"{dim_name} AS {self._quote_alias(output_name)}")

        # Add metrics
        for metric_ref in metrics:
            # Get metric name without model prefix
            if "." in metric_ref:
                metric_name = metric_ref.split(".", 1)[1]
            else:
                metric_name = metric_ref

            metric = model.get_metric(metric_name)
            if not metric:
                continue

            output_name = metric_output_name(metric_ref, metric_name)
            canonical_ref = metric_ref if "." in metric_ref else f"{model.name}.{metric_ref}"
            register_output_name(output_name, metric_ref, canonical_ref, metric_name, output_name)

            metric_expr = self._preaggregation_metric_expression(model, preagg, metric_name)
            select_exprs.append(f"{metric_expr} AS {self._quote_alias(output_name)}")

        # Build FROM clause
        from_clause = preagg_table

        # Build WHERE clause if filters exist
        where_clause = ""
        if filters:
            # Need to rewrite filters to use pre-agg column names
            rewritten_filters = []
            for f in filters:
                # Replace model_cte. with nothing (pre-agg table doesn't use CTEs)
                # Also replace model. with nothing
                rewritten_f = f.replace(f"{model.name}_cte.", "").replace(f"{model.name}.", "")

                # If this filter references the time dimension, map it to the pre-agg time column
                # e.g., created_at -> created_at_day for daily pre-agg
                if preagg.time_dimension and preagg.granularity:
                    time_col_name = f"{preagg.time_dimension}_{preagg.granularity}"
                    # Replace time dimension name with time column name
                    import re

                    # Match time dimension as a whole word (not part of another word)
                    rewritten_f = re.sub(r"\b" + re.escape(preagg.time_dimension) + r"\b", time_col_name, rewritten_f)

                rewritten_filters.append(rewritten_f)

            where_clause = f"\nWHERE {' AND '.join(rewritten_filters)}"

        # Build GROUP BY clause
        group_by_exprs = []
        for i in range(1, len(parsed_dims) + 1):
            group_by_exprs.append(str(i))

        group_by_clause = ""
        if group_by_exprs:
            group_by_clause = f"\nGROUP BY {', '.join(group_by_exprs)}"

        # Build HAVING clause for aggregate-stage metric filters
        having_clause = ""
        if aggregate_filters:
            rewritten_having_filters = [
                self._rewrite_preaggregation_aggregate_filter(model, preagg, f) for f in aggregate_filters
            ]
            having_clause = f"\nHAVING {' AND '.join(rewritten_having_filters)}"

        # Build ORDER BY clause
        order_by_clause = ""
        if order_by:
            order_clauses = []
            for field in order_by:
                parts = field.rsplit(" ", 1)
                direction = ""
                field_ref = field
                if len(parts) == 2 and parts[1].upper() in {"ASC", "DESC"}:
                    field_ref = parts[0]
                    direction = f" {parts[1].upper()}"

                field_name = field_ref.split(".", 1)[1] if "." in field_ref else field_ref
                output_name = output_names.get(field_ref) or output_names.get(field_name) or field_ref
                order_clauses.append(f"{self._quote_alias(output_name)}{direction}")
            order_by_clause = f"\nORDER BY {', '.join(order_clauses)}"

        # Build LIMIT/OFFSET clause
        limit_clause = ""
        if limit is not None:
            limit_clause = f"\nLIMIT {limit}"
        if offset is not None:
            limit_clause += f"\nOFFSET {offset}"

        # Combine into final query
        select_exprs_str = ",\n  ".join(select_exprs)
        query = f"""SELECT
  {select_exprs_str}
FROM {from_clause}{where_clause}{group_by_clause}{having_clause}{order_by_clause}{limit_clause}"""

        return query

    def _generate_instrumentation_comment(
        self, models: list[str], metrics: list[str], dimensions: list[str], used_preagg: bool = False
    ) -> str:
        """Generate instrumentation comment for query analysis.

        Args:
            models: List of model names used in query
            metrics: List of metric references
            dimensions: List of dimension references
            used_preagg: Whether this query used a pre-aggregation

        Returns:
            SQL comment string with metadata
        """
        # Extract granularities from dimensions
        granularities = set()
        clean_dims = []
        for dim in dimensions:
            if "__" in dim:
                parts = dim.rsplit("__", 1)
                clean_dims.append(parts[0])
                granularities.add(parts[1])
            else:
                clean_dims.append(dim)

        # Build metadata
        parts = []
        parts.append(f"models={','.join(sorted(models))}")
        parts.append(f"metrics={','.join(sorted(metrics))}")
        parts.append(f"dimensions={','.join(sorted(clean_dims))}")

        if granularities:
            parts.append(f"granularities={','.join(sorted(granularities))}")

        if used_preagg:
            parts.append("used_preagg=true")

        return f"-- sidemantic: {' '.join(parts)}"
