---
phase: 47-semi-additive-metrics
plan: 02
type: execute
wave: 2
depends_on: [47-01]
files_modified:
  - src/expand/mod.rs
  - src/expand/semi_additive.rs
  - src/expand/sql_gen.rs
  - src/expand/fan_trap.rs
  - test/sql/phase47_semi_additive.test
autonomous: true
requirements: [SEMI-01, SEMI-02, SEMI-03, SEMI-04, SEMI-05]
must_haves:
  truths:
    - "Querying a semi-additive metric produces correct snapshot-selected results via ROW_NUMBER CTE"
    - "A query with only regular metrics produces the same SQL as before (no CTE wrapping)"
    - "A query mixing regular and semi-additive metrics produces correct results for both"
    - "Semi-additive metrics are skipped by fan trap detection (no false positives)"
    - "Semi-additive metrics work correctly with multi-table JOINs"
    - "Semi-additive metrics work correctly with USING RELATIONSHIPS (scoped aliases)"
    - "When ALL non-additive dimensions are in the queried dimensions, the metric acts as regular additive (no CTE generated)"
    - "When the non-additive dimension is NOT in the query dimensions, ROW_NUMBER partitions by all queried dims and orders by NA dims"
  artifacts:
    - path: "src/expand/semi_additive.rs"
      provides: "CTE-based snapshot selection SQL generation"
      contains: "ROW_NUMBER"
    - path: "src/expand/sql_gen.rs"
      provides: "Dispatch to semi-additive expansion when any metric has active non_additive_by"
      contains: "has_active_semi_additive"
    - path: "src/expand/fan_trap.rs"
      provides: "Skip fan trap check for semi-additive metrics"
      contains: "non_additive_by"
    - path: "test/sql/phase47_semi_additive.test"
      provides: "End-to-end sqllogictest for semi-additive queries"
      contains: "NON ADDITIVE BY"
  key_links:
    - from: "src/expand/sql_gen.rs"
      to: "src/expand/semi_additive.rs"
      via: "expand() dispatches to expand_semi_additive when any metric has active (not effectively-regular) non_additive_by"
      pattern: "expand_semi_additive"
    - from: "src/expand/fan_trap.rs"
      to: "src/model.rs"
      via: "check_fan_traps filters out metrics with non-empty non_additive_by"
      pattern: "non_additive_by"
    - from: "src/expand/semi_additive.rs"
      to: "src/expand/sql_gen.rs"
      via: "Uses same resolution/join infrastructure (resolved_dims, resolved_exprs, etc.)"
      pattern: "resolve_joins_pkfk"
---

<objective>
Implement the CTE-based semi-additive expansion pipeline, fan trap interaction, and end-to-end integration tests.

Purpose: Makes semi-additive metrics actually queryable -- the model types from Plan 01 become functional at query time. This is the core expansion engine change.
Output: New semi_additive.rs module, expand() dispatch logic, fan trap exclusion, comprehensive sqllogictest coverage.
</objective>

<execution_context>
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
@$HOME/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/47-semi-additive-metrics/47-RESEARCH.md
@.planning/phases/47-semi-additive-metrics/47-01-SUMMARY.md

<interfaces>
<!-- Key types from Plan 01 and existing codebase. -->

From src/model.rs (created in Plan 01):
```rust
pub enum SortOrder { Asc, Desc }
pub enum NullsOrder { Last, First }
pub struct NonAdditiveDim {
    pub dimension: String,
    pub order: SortOrder,
    pub nulls: NullsOrder,
}
// Metric.non_additive_by: Vec<NonAdditiveDim>
```

From src/expand/sql_gen.rs (expand function, lines 215-428):
```rust
pub fn expand(
    view_name: &str,
    def: &SemanticViewDefinition,
    req: &QueryRequest,
) -> Result<String, ExpandError>
```
- Steps: validate request -> resolve dims -> resolve metrics -> inline_derived_metrics -> check_fan_traps -> find_using_context -> build SELECT -> FROM -> JOINs -> GROUP BY

From src/expand/types.rs:
```rust
pub struct QueryRequest {
    pub facts: Vec<String>,
    pub dimensions: Vec<String>,
    pub metrics: Vec<String>,
}
```

From src/expand/fan_trap.rs:
```rust
pub(super) fn check_fan_traps(
    view_name: &str,
    def: &SemanticViewDefinition,
    resolved_dims: &[&crate::model::Dimension],
    resolved_mets: &[&crate::model::Metric],
) -> Result<(), ExpandError>
```

From src/expand/resolution.rs:
```rust
pub fn quote_ident(name: &str) -> String;
pub fn quote_table_ref(table: &str) -> String;
```

From src/expand/join_resolver.rs:
```rust
pub(super) fn resolve_joins_pkfk(
    def: &SemanticViewDefinition,
    resolved_dims: &[&Dimension],
    resolved_mets: &[&Metric],
) -> Vec<String>;
pub(super) fn synthesize_on_clause(join: &Join, tables: &[TableRef]) -> String;
pub(super) fn synthesize_on_clause_scoped(join: &Join, tables: &[TableRef], scoped_alias: &str) -> String;
```

From src/expand/facts.rs:
```rust
pub(super) fn inline_derived_metrics(
    metrics: &[Metric],
    facts: &[Fact],
    topo_order: &[usize],
) -> HashMap<String, String>;
pub(super) fn toposort_facts(facts: &[Fact]) -> Option<Vec<usize>>;
```

From src/expand/role_playing.rs:
```rust
pub(super) fn find_using_context(
    view_name: &str,
    def: &SemanticViewDefinition,
    dim: &Dimension,
    resolved_mets: &[&Metric],
) -> Result<Option<String>, ExpandError>;
```

From src/util.rs:
```rust
pub fn replace_word_boundary(text: &str, from: &str, to: &str) -> String;
```

From src/expand/test_helpers.rs (created in Plan 01):
```rust
fn with_non_additive_by(self, metric_name: &str, dims: &[(&str, SortOrder, NullsOrder)]) -> Self;
```
</interfaces>
</context>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Semi-additive expansion module and expand() dispatch</name>
  <files>src/expand/semi_additive.rs, src/expand/mod.rs, src/expand/sql_gen.rs, src/expand/fan_trap.rs</files>
  <read_first>
    - src/expand/sql_gen.rs (full file -- understand the complete expand() function flow and SQL generation patterns)
    - src/expand/fan_trap.rs (full file -- understand check_fan_traps to add semi-additive skip)
    - src/expand/mod.rs (full file -- understand module declarations)
    - src/expand/join_resolver.rs (lines 1-30 -- understand resolve_joins_pkfk signature)
    - src/expand/role_playing.rs (lines 1-20 -- understand find_using_context signature)
    - src/expand/facts.rs (lines 1-30 -- understand inline_derived_metrics signature)
    - src/expand/resolution.rs (lines 1-20 -- understand quote_ident, find_dimension, find_metric)
    - src/model.rs (NonAdditiveDim, SortOrder, NullsOrder types from Plan 01)
    - src/expand/test_helpers.rs (with_non_additive_by builder from Plan 01)
  </read_first>
  <behavior>
    - Test: expand() with a semi-additive metric (NA dim NOT in query) generates SQL with `WITH __sv_snapshot AS (... ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) AS __sv_rn ...) SELECT ... SUM(CASE WHEN __sv_rn = 1 THEN ... END) ... FROM __sv_snapshot ... GROUP BY`
    - Test: expand() with only regular metrics (no non_additive_by) produces identical SQL as before (no CTE)
    - Test: expand() with mixed regular + semi-additive metrics on same table produces single CTE with conditional aggregation (`SUM(CASE WHEN __sv_rn = 1 THEN ... END)` for semi-additive, plain `SUM(...)` for regular)
    - Test: expand() with semi-additive metric where ALL NA dims ARE in query dims -> metric treated as effectively regular, NO CTE generated (Snowflake semantics)
    - Test: expand() with semi-additive metric where NA dim is NOT in query dims -> CTE with ROW_NUMBER PARTITION BY queried dims, ORDER BY NA dims
    - Test: check_fan_traps skips metrics with non-empty non_additive_by (no FanTrap error)
    - Test: expand() with semi-additive metric on multi-table JOIN -> FROM/JOIN clause in CTE includes all needed tables
    - Test: expand() with semi-additive metric + USING relationships -> scoped aliases in CTE
    - Test: expand() with metrics-only (no dims) semi-additive -> global aggregate with CTE
    - Test: expand() with DESC NULLS FIRST in non_additive_by -> ORDER BY has "DESC NULLS FIRST"
  </behavior>
  <action>
1. Create src/expand/semi_additive.rs as a new module:

```rust
//! CTE-based expansion for semi-additive metrics.
//!
//! When any requested metric has a non-empty `non_additive_by` field AND at least
//! one of its NA dims is NOT in the queried dimensions, the expansion wraps the
//! base query in a CTE that uses ROW_NUMBER() to select snapshot rows before
//! aggregation.
//!
//! **Effectively-regular classification (Snowflake semantics):**
//! A semi-additive metric whose ALL NA dims are present in the queried dimensions
//! is treated as "effectively regular" for that query -- no CTE needed, standard
//! aggregation applies. This matches Snowflake's behavior: "When the non-additive
//! dimension is included in the query, the metric is calculated as a standard
//! additive metric."
//!
//! The strategy for active semi-additive metrics:
//! - Single CTE (__sv_snapshot) containing all raw columns needed
//! - ROW_NUMBER() OVER (PARTITION BY non-NA-dims ORDER BY NA-dims) AS __sv_rn
//! - Outer SELECT: regular/effectively-regular metrics use plain aggregation,
//!   active semi-additive metrics use SUM(CASE WHEN __sv_rn = 1 THEN raw_val END)
//!
//! When multiple semi-additive metrics have different NON ADDITIVE BY dimensions,
//! each gets its own __sv_rn_N column in the CTE.

use std::collections::{HashMap, HashSet};

use crate::model::{Dimension, Metric, NonAdditiveDim, NullsOrder, SemanticViewDefinition, SortOrder};
use crate::util::replace_word_boundary;

use super::join_resolver::{resolve_joins_pkfk, synthesize_on_clause, synthesize_on_clause_scoped};
use super::resolution::{quote_ident, quote_table_ref};
use super::types::ExpandError;
```

The core public function classifies each metric into one of three categories:
- `regular`: no non_additive_by -> aggregate normally
- `effectively_regular`: has non_additive_by but ALL NA dims are in queried dims -> aggregate normally (Snowflake semantics)
- `active_semi_additive`: has at least one NA dim NOT in queried dims -> CASE WHEN __sv_rn = 1

```rust
/// Generate CTE-based expansion SQL for queries containing semi-additive metrics.
///
/// Called from `expand()` when `has_active_semi_additive` is true.
/// Receives already-resolved dims, metrics, expressions, and scoped aliases.
#[allow(clippy::too_many_lines, clippy::result_large_err)]
pub(super) fn expand_semi_additive(
    view_name: &str,
    def: &SemanticViewDefinition,
    resolved_dims: &[&Dimension],
    resolved_mets: &[&Metric],
    resolved_exprs: &HashMap<String, String>,
    dim_scoped_aliases: &[Option<String>],
) -> Result<String, ExpandError> {
    let mut sql = String::with_capacity(512);

    // Build set of queried dimension names for classification
    let queried_dim_names: HashSet<String> = resolved_dims
        .iter()
        .map(|d| d.name.to_ascii_lowercase())
        .collect();

    // Classify each metric
    let is_active_semi = |met: &Metric| -> bool {
        !met.non_additive_by.is_empty()
            && met.non_additive_by.iter().any(|na| {
                !queried_dim_names.contains(&na.dimension.to_ascii_lowercase())
            })
    };

    // 1. Identify distinct NON ADDITIVE BY dimension sets for ACTIVE metrics only.
    let na_groups = collect_na_groups(resolved_mets, &queried_dim_names);

    // === CTE ===
    sql.push_str("WITH __sv_snapshot AS (\n    SELECT\n");

    let mut cte_select_items: Vec<String> = Vec::new();

    // Dimension columns in CTE
    for (i, dim) in resolved_dims.iter().enumerate() {
        let mut base_expr = dim.expr.clone();
        if let Some(ref scoped) = dim_scoped_aliases[i] {
            if let Some(ref st) = dim.source_table {
                base_expr = replace_word_boundary(&base_expr, st, scoped);
            }
        }
        let final_expr = if let Some(ref type_str) = dim.output_type {
            format!("CAST({base_expr} AS {type_str})")
        } else {
            base_expr
        };
        cte_select_items.push(format!("        {} AS {}", final_expr, quote_ident(&dim.name)));
    }

    // Metric raw columns in CTE
    for (met_idx, met) in resolved_mets.iter().enumerate() {
        let resolved_expr = resolved_exprs
            .get(&met.name.to_ascii_lowercase())
            .cloned()
            .unwrap_or_else(|| met.expr.clone());

        if is_active_semi(met) {
            // Active semi-additive: extract inner expression from aggregate
            let inner = extract_aggregate_inner(&resolved_expr)
                .unwrap_or_else(|| resolved_expr.clone());
            cte_select_items.push(format!(
                "        {} AS \"__sv_semi_{}\"",
                inner, met_idx
            ));
        } else {
            // Regular or effectively-regular: include raw column reference
            let inner = extract_aggregate_inner(&resolved_expr)
                .unwrap_or_else(|| resolved_expr.clone());
            cte_select_items.push(format!(
                "        {} AS \"__sv_reg_{}\"",
                inner, met_idx
            ));
        }
    }

    // ROW_NUMBER columns (one per active NA group)
    for (group_idx, (na_dims, _met_indices)) in na_groups.iter().enumerate() {
        let rn_alias = if na_groups.len() == 1 {
            "__sv_rn".to_string()
        } else {
            format!("__sv_rn_{}", group_idx + 1)
        };

        // PARTITION BY: all queried dims EXCEPT the NA dims that are active
        // (NA dims not in query are not in resolved_dims, so this naturally
        // partitions by all queried dims)
        let na_dim_names: Vec<String> = na_dims
            .iter()
            .map(|nd| nd.dimension.to_ascii_lowercase())
            .collect();

        let partition_dims: Vec<String> = resolved_dims
            .iter()
            .filter(|d| !na_dim_names.contains(&d.name.to_ascii_lowercase()))
            .map(|d| quote_ident(&d.name))
            .collect();

        let partition_clause = if partition_dims.is_empty() {
            String::new()
        } else {
            format!("PARTITION BY {}", partition_dims.join(", "))
        };

        // ORDER BY: the NA dims with their specified sort order
        // Use the dimension's raw expression (not CTE alias) since the NA dim
        // may not be in the queried dimensions and thus not in the CTE SELECT
        let order_items: Vec<String> = na_dims
            .iter()
            .map(|nd| {
                // Find the dimension definition to get its expression
                let dim_expr = resolved_dims
                    .iter()
                    .find(|d| d.name.eq_ignore_ascii_case(&nd.dimension))
                    .map(|d| quote_ident(&d.name))
                    .unwrap_or_else(|| {
                        // NA dim not in queried dims -- find it in the view definition
                        // and use its raw expression
                        def.dimensions
                            .iter()
                            .find(|d| d.name.eq_ignore_ascii_case(&nd.dimension))
                            .map(|d| d.expr.clone())
                            .unwrap_or_else(|| quote_ident(&nd.dimension))
                    });
                let dir = match nd.order {
                    SortOrder::Asc => "ASC",
                    SortOrder::Desc => "DESC",
                };
                let nulls = match nd.nulls {
                    NullsOrder::First => "NULLS FIRST",
                    NullsOrder::Last => "NULLS LAST",
                };
                format!("{dim_expr} {dir} {nulls}")
            })
            .collect();

        let order_clause = order_items.join(", ");

        let window_spec = if partition_clause.is_empty() {
            format!("ORDER BY {order_clause}")
        } else {
            format!("{partition_clause} ORDER BY {order_clause}")
        };

        cte_select_items.push(format!(
            "        ROW_NUMBER() OVER ({window_spec}) AS \"{rn_alias}\""
        ));
    }

    sql.push_str(&cte_select_items.join(",\n"));

    // CTE FROM clause (same logic as expand())
    sql.push_str("\n    FROM ");
    sql.push_str(&quote_table_ref(&def.base_table));
    if let Some(base_ref) = def.tables.first() {
        sql.push_str(" AS ");
        sql.push_str(&quote_ident(&base_ref.alias));
    }

    // CTE JOINs
    let ordered_aliases = resolve_joins_pkfk(def, resolved_dims, resolved_mets);
    for alias in &ordered_aliases {
        if let Some(sep_pos) = alias.find("__") {
            let rel_name = &alias[sep_pos + 2..];
            let Some(join) = def.joins.iter().find(|j| {
                j.name.as_ref().is_some_and(|n| n.eq_ignore_ascii_case(rel_name))
            }) else {
                continue;
            };
            let bare_alias = &alias[..sep_pos];
            let table_ref = def.tables.iter().find(|t| t.alias.to_ascii_lowercase() == bare_alias);
            let physical_table = table_ref.map_or(bare_alias, |t| t.table.as_str());
            sql.push_str("\n    LEFT JOIN ");
            sql.push_str(&quote_table_ref(physical_table));
            sql.push_str(" AS ");
            sql.push_str(&quote_ident(alias));
            sql.push_str(" ON ");
            sql.push_str(&synthesize_on_clause_scoped(join, &def.tables, alias));
        } else {
            let Some(join) = def.joins.iter().find(|j| {
                j.table.to_ascii_lowercase() == *alias || j.from_alias.to_ascii_lowercase() == *alias
            }) else {
                continue;
            };
            let table_ref = def.tables.iter().find(|t| t.alias.to_ascii_lowercase() == *alias);
            let physical_table = table_ref.map_or(alias.as_str(), |t| t.table.as_str());
            sql.push_str("\n    LEFT JOIN ");
            sql.push_str(&quote_table_ref(physical_table));
            sql.push_str(" AS ");
            sql.push_str(&quote_ident(alias));
            sql.push_str(" ON ");
            sql.push_str(&synthesize_on_clause(join, &def.tables));
        }
    }

    sql.push_str("\n)\n");

    // === Outer SELECT ===
    sql.push_str("SELECT\n");

    let mut outer_select_items: Vec<String> = Vec::new();

    // Dimension columns: reference CTE aliases
    for dim in resolved_dims.iter() {
        outer_select_items.push(format!("    {} AS {}", quote_ident(&dim.name), quote_ident(&dim.name)));
    }

    // Metric columns
    for (met_idx, met) in resolved_mets.iter().enumerate() {
        let resolved_expr = resolved_exprs
            .get(&met.name.to_ascii_lowercase())
            .cloned()
            .unwrap_or_else(|| met.expr.clone());
        let agg_func = extract_aggregate_func(&resolved_expr).unwrap_or("SUM");

        if is_active_semi(met) {
            // Active semi-additive: wrap in CASE WHEN __sv_rn = 1
            let rn_col = get_rn_column_for_metric(met_idx, &na_groups);
            let final_expr = format!(
                "{agg_func}(CASE WHEN \"{rn_col}\" = 1 THEN \"__sv_semi_{met_idx}\" END)"
            );
            let cast_expr = if let Some(ref type_str) = met.output_type {
                format!("CAST({final_expr} AS {type_str})")
            } else {
                final_expr
            };
            outer_select_items.push(format!("    {} AS {}", cast_expr, quote_ident(&met.name)));
        } else {
            // Regular or effectively-regular: aggregate normally over all rows
            let final_expr = format!("{agg_func}(\"__sv_reg_{met_idx}\")");
            let cast_expr = if let Some(ref type_str) = met.output_type {
                format!("CAST({final_expr} AS {type_str})")
            } else {
                final_expr
            };
            outer_select_items.push(format!("    {} AS {}", cast_expr, quote_ident(&met.name)));
        }
    }

    sql.push_str(&outer_select_items.join(",\n"));

    // FROM the CTE
    sql.push_str("\nFROM __sv_snapshot");

    // GROUP BY (when dims are present)
    if !resolved_dims.is_empty() {
        sql.push_str("\nGROUP BY\n");
        let group_items: Vec<String> = (1..=resolved_dims.len())
            .map(|i| format!("    {i}"))
            .collect();
        sql.push_str(&group_items.join(",\n"));
    }

    Ok(sql)
}
```

Helper functions in semi_additive.rs:

```rust
/// Group metrics by their NON ADDITIVE BY dimension sets.
/// Only includes ACTIVE semi-additive metrics (those with at least one NA dim
/// not in the queried dimensions).
/// Returns Vec of (na_dims, metric_indices_with_this_set).
fn collect_na_groups(
    resolved_mets: &[&Metric],
    queried_dim_names: &HashSet<String>,
) -> Vec<(Vec<NonAdditiveDim>, Vec<usize>)> {
    let mut groups: Vec<(Vec<String>, Vec<NonAdditiveDim>, Vec<usize>)> = Vec::new();
    for (idx, met) in resolved_mets.iter().enumerate() {
        if met.non_additive_by.is_empty() {
            continue;
        }
        // Skip effectively-regular metrics (all NA dims in query)
        let all_na_in_query = met.non_additive_by.iter().all(|na| {
            queried_dim_names.contains(&na.dimension.to_ascii_lowercase())
        });
        if all_na_in_query {
            continue;
        }
        let key: Vec<String> = met.non_additive_by
            .iter()
            .map(|nd| nd.dimension.to_ascii_lowercase())
            .collect();
        if let Some(group) = groups.iter_mut().find(|(k, _, _)| *k == key) {
            group.2.push(idx);
        } else {
            groups.push((key, met.non_additive_by.clone(), vec![idx]));
        }
    }
    groups.into_iter().map(|(_, dims, indices)| (dims, indices)).collect()
}

/// Extract the inner expression from an aggregate function call.
/// e.g., "SUM(a.balance)" -> Some("a.balance"), "COUNT(*)" -> Some("*"),
/// "revenue - cost" -> None (not an aggregate).
fn extract_aggregate_inner(expr: &str) -> Option<String> {
    let trimmed = expr.trim();
    let open = trimmed.find('(')?;
    let close = trimmed.rfind(')')?;
    if close <= open {
        return None;
    }
    // Verify the part before '(' looks like a function name (alphanumeric/underscore)
    let func_name = trimmed[..open].trim();
    if func_name.is_empty() || !func_name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
        return None;
    }
    Some(trimmed[open + 1..close].trim().to_string())
}

/// Extract the aggregate function name from an expression.
/// e.g., "SUM(a.balance)" -> Some("SUM"), "COUNT(*)" -> Some("COUNT").
fn extract_aggregate_func(expr: &str) -> Option<&str> {
    let trimmed = expr.trim();
    let open = trimmed.find('(')?;
    let func_name = trimmed[..open].trim();
    if func_name.is_empty() || !func_name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
        return None;
    }
    Some(func_name)
}

/// Get the ROW_NUMBER column name for a given metric index.
fn get_rn_column_for_metric(met_idx: usize, na_groups: &[(Vec<NonAdditiveDim>, Vec<usize>)]) -> String {
    if na_groups.len() == 1 {
        return "__sv_rn".to_string();
    }
    for (group_idx, (_dims, indices)) in na_groups.iter().enumerate() {
        if indices.contains(&met_idx) {
            return format!("__sv_rn_{}", group_idx + 1);
        }
    }
    "__sv_rn".to_string() // fallback
}
```

2. In src/expand/mod.rs, add the semi_additive module declaration:
```rust
mod semi_additive;
```
Add it after `mod role_playing;` and before `mod sql_gen;`.

3. In src/expand/sql_gen.rs, modify the `expand()` function to dispatch to semi-additive path.

After the fan trap check (line ~300) and scoped aliases computation (line ~308), add a check:
```rust
// Phase 47: Check if any resolved metric ACTUALLY needs semi-additive expansion.
// A semi-additive metric only needs CTE treatment when at least one of its
// NA dims is NOT in the queried dimension set. When ALL NA dims are in the
// query, the metric acts as regular (Snowflake semantics).
let queried_dim_names: std::collections::HashSet<String> = resolved_dims
    .iter()
    .map(|d| d.name.to_ascii_lowercase())
    .collect();
let has_active_semi_additive = resolved_mets.iter().any(|m| {
    !m.non_additive_by.is_empty()
        && m.non_additive_by.iter().any(|na| {
            !queried_dim_names.contains(&na.dimension.to_ascii_lowercase())
        })
});

if has_active_semi_additive {
    return super::semi_additive::expand_semi_additive(
        view_name,
        def,
        &resolved_dims,
        &resolved_mets,
        &resolved_exprs,
        &dim_scoped_aliases,
    );
}
```

This goes right before step 5 ("Build the SELECT clause") at line ~310. The existing code path remains untouched for regular-only queries AND for queries where all semi-additive metrics are effectively regular.

4. In src/expand/fan_trap.rs, modify `check_fan_traps` to skip semi-additive metrics.

In the loop `for met in resolved_mets {` (line ~61), add at the top of the loop body:
```rust
// Phase 47: Skip fan trap check for semi-additive metrics.
// The ROW_NUMBER CTE inherently handles fan-out by selecting one row
// per partition, making fan trap detection unnecessary.
if !met.non_additive_by.is_empty() {
    continue;
}
```

5. Add comprehensive unit tests in src/expand/semi_additive.rs:

Test cases in a `#[cfg(test)] mod tests` block:

a. `test_semi_additive_single_metric_single_dim`: Single table, one semi-additive metric `SUM(balance) NON ADDITIVE BY (date_dim DESC)`, dim `customer_id` (date_dim NOT in query). Verify SQL contains `WITH __sv_snapshot AS`, `ROW_NUMBER() OVER (PARTITION BY "customer_id" ORDER BY ... DESC NULLS FIRST)`, `SUM(CASE WHEN "__sv_rn" = 1 THEN "__sv_semi_0" END)`.

b. `test_effectively_regular_all_na_dims_in_query`: Semi-additive metric with NA dim `date_dim`, query dims `[customer_id, date_dim]`. Since ALL NA dims are in the query, `has_active_semi_additive` is false -> expand() uses the regular path (no CTE). Verify output does NOT contain `WITH __sv_snapshot`.

c. `test_semi_additive_na_dim_not_in_query`: Query with dim `customer_id` only, NA dim `date_dim` not requested. Verify PARTITION BY includes `customer_id`, ORDER BY uses `date_dim` expression.

d. `test_mixed_regular_and_semi_additive`: Two metrics: `SUM(amount)` (regular) + `SUM(balance) NON ADDITIVE BY (date_dim DESC)` (date_dim NOT in query). Verify CTE includes both `__sv_reg_0` and `__sv_semi_1`, outer SELECT uses plain `SUM("__sv_reg_0")` for regular and `SUM(CASE WHEN "__sv_rn" = 1 THEN "__sv_semi_1" END)` for semi-additive.

e. `test_no_semi_additive_no_cte`: Regular-only metrics -> expand() produces flat SELECT (no WITH).

f. `test_extract_aggregate_inner`: Unit test for the helper function.

g. `test_metrics_only_global_aggregate`: No dims, one semi-additive metric. Verify ROW_NUMBER PARTITION BY is empty (global), outer SELECT has no GROUP BY.

h. `test_fan_trap_skips_semi_additive`: Multi-table view with fan-out relationship, semi-additive metric on fan-out table -> no FanTrap error.

6. Add tests for multi-table JOIN in semi_additive.rs tests:
- Two tables with PK/FK, semi-additive metric on joined table -> CTE includes LEFT JOIN
- Semi-additive with USING relationship -> CTE uses scoped alias

  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test semi_additive 2>&1 | tail -10 && cargo test fan_trap 2>&1 | tail -5 && cargo test expand 2>&1 | tail -5</automated>
  </verify>
  <acceptance_criteria>
    - src/expand/semi_additive.rs exists and contains `pub(super) fn expand_semi_additive`
    - src/expand/semi_additive.rs contains `ROW_NUMBER() OVER`
    - src/expand/semi_additive.rs contains `CASE WHEN` for conditional aggregation
    - src/expand/semi_additive.rs contains `fn extract_aggregate_inner`
    - src/expand/semi_additive.rs contains `fn collect_na_groups` that takes `queried_dim_names` parameter
    - src/expand/semi_additive.rs `collect_na_groups` skips effectively-regular metrics (all NA dims in query)
    - src/expand/mod.rs contains `mod semi_additive;`
    - src/expand/sql_gen.rs contains `has_active_semi_additive` check (not just `has_semi_additive`)
    - src/expand/sql_gen.rs classification uses `queried_dim_names` to determine active vs effectively-regular
    - src/expand/fan_trap.rs contains `non_additive_by.is_empty()` skip check
    - `cargo test semi_additive` exits 0
    - `cargo test fan_trap` exits 0
    - `cargo test expand` exits 0 (all existing expand tests still pass)
  </acceptance_criteria>
  <done>Semi-additive expansion produces CTE-based SQL with ROW_NUMBER snapshot selection. Metrics with all NA dims in query are treated as effectively regular (no CTE). Mixed regular+semi-additive queries use conditional aggregation. Fan trap skips semi-additive metrics. Multi-table JOINs and USING work. All unit tests pass.</done>
</task>

<task type="auto">
  <name>Task 2: End-to-end sqllogictest integration tests</name>
  <files>test/sql/phase47_semi_additive.test</files>
  <read_first>
    - test/sql/phase46_fact_query.test (recent sqllogictest example -- understand test pattern, CREATE TABLE, CREATE SEMANTIC VIEW, INSERT, query, verify)
    - test/sql/phase31_fan_trap.test (fan trap test patterns for multi-table views)
    - test/sql/phase32_role_playing.test (USING RELATIONSHIPS test patterns)
    - test/sql/TEST_LIST (to add the new test file)
    - src/expand/semi_additive.rs (from Task 1 -- understand generated SQL format)
  </read_first>
  <action>
1. Create test/sql/phase47_semi_additive.test with the following test cases.

**Important design notes for all tests:**
- When the NA dim IS in the query dims, the metric acts as regular (no CTE). This means all rows return with their actual values.
- When the NA dim is NOT in the query dims, snapshot selection applies via ROW_NUMBER CTE.
- Test data must be deterministic: avoid ties in the ORDER BY column within a partition.

```
# Phase 47: Semi-additive metrics

require semantic_views

# ========================================
# Setup: accounts + customers tables
# ========================================

statement ok
CREATE TABLE customers (id INTEGER PRIMARY KEY, name VARCHAR);

statement ok
INSERT INTO customers VALUES (1, 'Alice'), (2, 'Bob');

statement ok
CREATE TABLE accounts (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER,
    report_date DATE,
    balance DECIMAL(10,2)
);

statement ok
INSERT INTO accounts VALUES
    (1, 1, '2024-01-01', 100.00),
    (2, 1, '2024-01-02', 150.00),
    (3, 1, '2024-01-03', 200.00),
    (4, 2, '2024-01-01', 300.00),
    (5, 2, '2024-01-02', 250.00);

# ========================================
# Test 1: Basic semi-additive metric DDL
# ========================================

statement ok
CREATE SEMANTIC VIEW account_view AS
TABLES (
    a AS accounts PRIMARY KEY (id),
    c AS customers PRIMARY KEY (id)
)
RELATIONSHIPS (
    a.customer_id -> c.id
)
DIMENSIONS (
    c.customer_name AS c.name,
    a.report_date AS a.report_date
)
METRICS (
    a.total_balance NON ADDITIVE BY (report_date DESC) AS SUM(a.balance)
)

# ========================================
# Test 2: Query semi-additive metric (NA dim NOT in query)
# Latest balance per customer (snapshot at most recent date)
# Alice: latest is 2024-01-03 -> 200.00
# Bob: latest is 2024-01-02 -> 250.00
# ========================================

query TR
SELECT * FROM semantic_view('account_view', dimensions := ['customer_name'], metrics := ['total_balance']) ORDER BY 1;
----
Alice	200.00
Bob	250.00

# ========================================
# Test 3: Semi-additive with NA dim IN query
# When report_date IS in dims, metric acts as regular
# (Snowflake semantics: no snapshot filtering when NA dim is queried)
# All rows returned with their actual balance values
# ========================================

query TDR
SELECT * FROM semantic_view('account_view', dimensions := ['customer_name', 'report_date'], metrics := ['total_balance']) ORDER BY 1, 2;
----
Alice	2024-01-01	100.00
Alice	2024-01-02	150.00
Alice	2024-01-03	200.00
Bob	2024-01-01	300.00
Bob	2024-01-02	250.00

# ========================================
# Test 4: NON ADDITIVE BY validation
# Reference nonexistent dimension -> error
# ========================================

statement error
CREATE SEMANTIC VIEW bad_na AS
TABLES (a AS accounts PRIMARY KEY (id))
DIMENSIONS (a.report_date AS a.report_date)
METRICS (a.total_balance NON ADDITIVE BY (nonexistent_dim) AS SUM(a.balance))
----
does not match any declared dimension

# ========================================
# Test 5: NON ADDITIVE BY on derived metric -> error
# ========================================

statement error
CREATE SEMANTIC VIEW bad_na2 AS
TABLES (a AS accounts PRIMARY KEY (id))
DIMENSIONS (a.report_date AS a.report_date)
METRICS (
    a.total_balance AS SUM(a.balance),
    adjusted NON ADDITIVE BY (report_date) AS total_balance * 1.1
)
----
NON ADDITIVE BY

# ========================================
# Test 6: Mixed regular + semi-additive metrics
# row_count: counts ALL rows per customer (regular)
# total_balance: latest snapshot only (semi-additive, report_date not in query)
# Alice: 3 rows, latest balance = 200.00
# Bob: 2 rows, latest balance = 250.00
# ========================================

statement ok
CREATE SEMANTIC VIEW mixed_view AS
TABLES (
    a AS accounts PRIMARY KEY (id),
    c AS customers PRIMARY KEY (id)
)
RELATIONSHIPS (
    a.customer_id -> c.id
)
DIMENSIONS (
    c.customer_name AS c.name,
    a.report_date AS a.report_date
)
METRICS (
    a.row_count AS COUNT(*),
    a.total_balance NON ADDITIVE BY (report_date DESC) AS SUM(a.balance)
)

query TIR
SELECT * FROM semantic_view('mixed_view', dimensions := ['customer_name'], metrics := ['row_count', 'total_balance']) ORDER BY 1;
----
Alice	3	200.00
Bob	2	250.00

# ========================================
# Test 7: GET_DDL round-trip with NON ADDITIVE BY
# ========================================

query I
SELECT GET_DDL('SEMANTIC_VIEW', 'account_view') LIKE '%NON ADDITIVE BY%';
----
true

# ========================================
# Test 8: DESCRIBE shows NON_ADDITIVE_BY
# ========================================

query TTTTT
SELECT * FROM TABLE(DESCRIBE_SEMANTIC_VIEW('account_view')) WHERE property = 'NON_ADDITIVE_BY';
----
METRIC	total_balance	accounts	NON_ADDITIVE_BY	report_date DESC

# ========================================
# Test 9: Fan trap NOT triggered for semi-additive
# (account -> customer is many-to-one, which is safe,
# but if metric were on the "one" side querying dim on "many" side,
# fan trap would fire for regular metrics but not semi-additive)
# ========================================

# Drop and recreate with fan-trap scenario
statement ok
DROP SEMANTIC VIEW IF EXISTS mixed_view;

statement ok
DROP SEMANTIC VIEW IF EXISTS account_view;

# ========================================
# Test 10: Global aggregate (no dims)
# Uses single-customer data to ensure deterministic result.
# With no partition and ORDER BY report_date DESC,
# rn=1 is the row with 2024-01-03 (balance=200.00).
# SUM of that single snapshot row = 200.00.
# ========================================

statement ok
CREATE TABLE single_acct (
    id INTEGER PRIMARY KEY,
    report_date DATE,
    balance DECIMAL(10,2)
);

statement ok
INSERT INTO single_acct VALUES
    (1, '2024-01-01', 100.00),
    (2, '2024-01-02', 150.00),
    (3, '2024-01-03', 200.00);

statement ok
CREATE SEMANTIC VIEW global_view AS
TABLES (a AS single_acct PRIMARY KEY (id))
DIMENSIONS (a.report_date AS a.report_date)
METRICS (
    a.latest_balance NON ADDITIVE BY (report_date DESC) AS SUM(a.balance)
)

query R
SELECT * FROM semantic_view('global_view', metrics := ['latest_balance']);
----
200.00

# ========================================
# Cleanup
# ========================================

statement ok
DROP SEMANTIC VIEW IF EXISTS global_view;

statement ok
DROP TABLE IF EXISTS single_acct;

statement ok
DROP TABLE IF EXISTS accounts;

statement ok
DROP TABLE IF EXISTS customers;
```

2. Add `phase47_semi_additive.test` to test/sql/TEST_LIST (append at the end or in sorted position).

3. Run `just build && just test-sql` to verify all sqllogictest tests pass, including the new phase47 tests.

  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && just build && just test-sql 2>&1 | tail -10</automated>
  </verify>
  <acceptance_criteria>
    - test/sql/phase47_semi_additive.test exists and contains at least 8 test cases
    - test/sql/phase47_semi_additive.test contains `NON ADDITIVE BY`
    - test/sql/phase47_semi_additive.test contains `CREATE SEMANTIC VIEW` with semi-additive metric
    - test/sql/phase47_semi_additive.test contains mixed regular + semi-additive query test
    - test/sql/phase47_semi_additive.test Test 3 expects ALL rows (effectively-regular when NA dim in query)
    - test/sql/phase47_semi_additive.test Test 10 uses deterministic single-entity data
    - test/sql/phase47_semi_additive.test contains define-time validation error test (nonexistent dim)
    - test/sql/phase47_semi_additive.test contains GET_DDL round-trip check
    - test/sql/phase47_semi_additive.test contains DESCRIBE NON_ADDITIVE_BY property check
    - test/sql/TEST_LIST contains `phase47_semi_additive.test`
    - `just build && just test-sql` exits 0
    - `just test-all` exits 0
  </acceptance_criteria>
  <done>End-to-end sqllogictest validates: semi-additive DDL parsing, snapshot query with correct values, effectively-regular behavior when NA dim is queried, mixed regular+semi-additive queries, deterministic global aggregate, define-time validation errors, GET_DDL round-trip, DESCRIBE surfacing. Full test suite passes.</done>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| Model -> CTE SQL generation | Non-additive dimension names embedded in ORDER BY and PARTITION BY clauses |
| Query request -> expansion dispatch | User-chosen metric names determine expansion path |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-47-04 | Tampering | semi_additive.rs ORDER BY clause | mitigate | All dimension references use `quote_ident()` which double-quotes identifiers, preventing SQL injection |
| T-47-05 | Tampering | semi_additive.rs CTE column aliases | mitigate | Internal aliases (__sv_rn, __sv_semi_N) are hardcoded, not user-controlled |
| T-47-06 | Information Disclosure | semi_additive.rs | accept | CTE column names (__sv_rn, __sv_semi) are visible in EXPLAIN output but contain no sensitive data |
| T-47-07 | Denial of Service | expand_semi_additive with many metrics | accept | DuckDB's own query planner handles complexity; CTE overhead is O(metrics * dims) |
</threat_model>

<verification>
- `cargo test` passes (all unit tests including semi_additive module, fan_trap, expand)
- `just build && just test-sql` passes (all sqllogictest including phase47_semi_additive.test)
- `just test-all` passes (full quality gate)
- Existing expand tests still pass (regular-only queries unchanged)
</verification>

<success_criteria>
- Semi-additive metrics produce correct snapshot-selected results via ROW_NUMBER CTE
- Regular-only queries produce identical SQL to before (no CTE overhead)
- Mixed regular + semi-additive queries produce correct results for both types
- When ALL NA dims are in the query, metric acts as effectively regular (no CTE, standard aggregation)
- When any NA dim is NOT in the query, CTE with ROW_NUMBER is generated
- Fan trap detection skips semi-additive metrics
- Multi-table JOINs preserved in CTE
- USING RELATIONSHIPS (scoped aliases) work in CTE
- All sqllogictest integration tests pass
- Full `just test-all` quality gate passes
</success_criteria>

<output>
After completion, create `.planning/phases/47-semi-additive-metrics/47-02-SUMMARY.md`
</output>
