---
phase: 48-window-function-metrics
plan: 02
type: execute
wave: 2
depends_on: [48-01]
files_modified:
  - src/expand/window.rs
  - src/expand/mod.rs
  - src/expand/sql_gen.rs
  - src/expand/fan_trap.rs
  - src/ddl/show_dims_for_metric.rs
  - test/sql/phase48_window_metrics.test
  - test/sql/TEST_LIST
autonomous: true
requirements: [WIN-02, WIN-03, WIN-04, WIN-05]
must_haves:
  truths:
    - "Querying a window function metric produces correct results via CTE aggregation + outer window SELECT"
    - "Attempting to mix window function metrics with aggregate metrics produces a clear blocking error"
    - "Window function metrics are excluded from fan trap detection"
    - "SHOW SEMANTIC DIMENSIONS FOR METRIC shows required=TRUE for dimensions in EXCLUDING and ORDER BY"
    - "Multiple window metrics in the same query each apply their own OVER clause correctly"
    - "Window metrics with frame clauses produce correct SQL (RANGE BETWEEN, ROWS BETWEEN)"
  artifacts:
    - path: "src/expand/window.rs"
      provides: "Window metric CTE expansion"
      contains: "expand_window_metrics"
    - path: "src/expand/sql_gen.rs"
      provides: "Window metric dispatch in expand()"
      contains: "has_window_metrics"
    - path: "src/expand/fan_trap.rs"
      provides: "Window metric skip in fan trap check"
      contains: "is_window"
    - path: "src/ddl/show_dims_for_metric.rs"
      provides: "required=TRUE for window metric dims"
      contains: "window_spec"
    - path: "test/sql/phase48_window_metrics.test"
      provides: "End-to-end sqllogictest integration"
  key_links:
    - from: "src/expand/sql_gen.rs"
      to: "src/expand/window.rs"
      via: "expand_window_metrics() dispatch"
      pattern: "expand_window_metrics"
    - from: "src/expand/window.rs"
      to: "src/model.rs"
      via: "WindowSpec field access for SQL generation"
      pattern: "window_spec"
    - from: "src/expand/fan_trap.rs"
      to: "src/model.rs"
      via: "is_window() check for skip"
      pattern: "is_window"
    - from: "src/ddl/show_dims_for_metric.rs"
      to: "src/model.rs"
      via: "WindowSpec excluding_dims + order_by for required flag"
      pattern: "window_spec"
---

<objective>
Implement the window metric expansion pipeline, window/aggregate mixing error, fan trap exclusion, SHOW DIMS required=TRUE, and end-to-end integration tests.

Purpose: Completes the window function metrics feature. After this plan, users can declare window metrics in DDL and query them with correct CTE-based SQL generation, partition-aware windowing, and proper error handling for invalid combinations.

Output: New src/expand/window.rs module, expand() dispatch for window metrics, fan trap skip, SHOW DIMS required column, phase48 sqllogictest file.
</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/48-window-function-metrics/48-RESEARCH.md
@.planning/phases/48-window-function-metrics/48-01-SUMMARY.md
@.planning/phases/47-semi-additive-metrics/47-02-SUMMARY.md

<interfaces>
<!-- Key types and contracts the executor needs from Plan 01 and existing code. -->

From src/model.rs (WindowSpec, added by Plan 01):
```rust
pub struct WindowSpec {
    pub window_function: String,
    pub inner_metric: String,
    pub extra_args: Vec<String>,
    pub excluding_dims: Vec<String>,
    pub order_by: Vec<WindowOrderBy>,
    pub frame_clause: Option<String>,
}

pub struct WindowOrderBy {
    pub expr: String,
    pub order: SortOrder,
    pub nulls: NullsOrder,
}

impl Metric {
    pub fn is_window(&self) -> bool { self.window_spec.is_some() }
}
```

From src/expand/sql_gen.rs (expand function signature and dispatch pattern):
```rust
pub fn expand(
    view_name: &str,
    def: &SemanticViewDefinition,
    req: &QueryRequest,
) -> Result<String, ExpandError>;

// Dispatch precedence in expand():
// 1. Facts + metrics mutual exclusion check
// 2. Empty request check
// 3. Facts dispatch -> expand_facts()
// 4. Resolve dims + metrics
// 5. inline_derived_metrics
// 6. check_fan_traps
// 7. Semi-additive dispatch -> expand_semi_additive()
// 8. Regular expansion
```

From src/expand/types.rs (ExpandError enum):
```rust
pub enum ExpandError {
    EmptyRequest { view_name: String },
    UnknownDimension { ... },
    UnknownMetric { ... },
    DuplicateDimension { ... },
    DuplicateMetric { ... },
    FanTrap { ... },
    AmbiguousPath { ... },
    PrivateMetric { ... },
    PrivateFact { ... },
    FactsMetricsMutualExclusion { ... },
    UnknownFact { ... },
    DuplicateFact { ... },
    FactPathViolation { ... },
}
```

From src/expand/semi_additive.rs (CTE pattern to follow):
```rust
pub(super) fn expand_semi_additive(
    view_name: &str,
    def: &SemanticViewDefinition,
    resolved_dims: &[&crate::model::Dimension],
    resolved_mets: &[&Metric],
    resolved_exprs: &HashMap<String, String>,
    dim_scoped_aliases: &[Option<String>],
) -> Result<String, ExpandError>;
```

From src/expand/fan_trap.rs (skip pattern):
```rust
// Line 65: Skip semi-additive
if !met.non_additive_by.is_empty() {
    continue;
}
```

From src/ddl/show_dims_for_metric.rs (current required column: constant FALSE):
```rust
struct ShowDimForMetricRow {
    table_name: String,
    name: String,
    data_type: String,
}
// In func(): req_slice[i] = false; // constant FALSE for all rows
```

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

From src/expand/join_resolver.rs:
```rust
pub(super) fn resolve_joins_pkfk(...) -> Vec<String>;
pub(super) fn synthesize_on_clause(...) -> String;
pub(super) fn synthesize_on_clause_scoped(...) -> String;
```
</interfaces>
</context>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Window metric expansion module, expand() dispatch, fan trap skip, and mixing error</name>
  <files>src/expand/window.rs, src/expand/mod.rs, src/expand/sql_gen.rs, src/expand/types.rs, src/expand/fan_trap.rs</files>
  <read_first>
    - src/expand/semi_additive.rs (CTE generation pattern, expand_semi_additive signature, extract_aggregate_inner helper)
    - src/expand/sql_gen.rs (full expand() function for dispatch pattern, resolve_joins_pkfk usage, dim_scoped_aliases)
    - src/expand/types.rs (ExpandError enum for new variant)
    - src/expand/fan_trap.rs (semi-additive skip pattern at line 65)
    - src/expand/mod.rs (module declarations, pub exports)
    - src/expand/resolution.rs (quote_ident, find_dimension, find_metric)
    - src/expand/join_resolver.rs (resolve_joins_pkfk signature, synthesize_on_clause)
    - src/model.rs (WindowSpec struct from Plan 01, Metric.is_window())
    - src/expand/test_helpers.rs (with_window_spec builder from Plan 01)
  </read_first>
  <behavior>
    - Test 1: expand() with a single window metric + 3 dims generates CTE with GROUP BY all dims + outer SELECT with window function and correct PARTITION BY (all queried dims minus excluded)
    - Test 2: expand() with a window metric excluding 2 dims generates PARTITION BY with only the remaining dims
    - Test 3: expand() with a window metric that has a frame clause includes the frame clause in the OVER
    - Test 4: expand() with a window metric using LAG(metric, 30) includes extra args in the function call
    - Test 5: expand() with two window metrics sharing the same CTE but different EXCLUDING sets generates correct PARTITION BY for each
    - Test 6: expand() mixing window metric + regular aggregate metric returns Err(WindowAggregateMixing)
    - Test 7: expand() with window metric where required dim is missing returns Err (new error variant)
    - Test 8: fan trap check skips window metrics (no FanTrap error for window metric across many-to-one boundary)
  </behavior>
  <action>
1. In src/expand/types.rs, add two new error variants to ExpandError:

```rust
    /// Window function metrics cannot be mixed with aggregate metrics.
    WindowAggregateMixing {
        view_name: String,
        window_metrics: Vec<String>,
        aggregate_metrics: Vec<String>,
    },
    /// A dimension required by a window metric (EXCLUDING or ORDER BY) is not in the query.
    WindowMetricRequiredDimension {
        view_name: String,
        metric_name: String,
        dimension_name: String,
        reason: String,  // "PARTITION BY EXCLUDING" or "ORDER BY"
    },
```

Add Display implementations:
- WindowAggregateMixing: `"semantic view '{view_name}': cannot mix window function metrics [{window_metrics joined}] with aggregate metrics [{aggregate_metrics joined}] in the same query"`
- WindowMetricRequiredDimension: `"semantic view '{view_name}': window function metric '{metric_name}' requires dimension '{dimension_name}' to be included in the query (used in {reason})"`

2. Create src/expand/window.rs:

```rust
//! CTE-based expansion for window function metrics.
//!
//! Window metrics wrap another metric in a SQL window function (e.g., AVG(total_qty)
//! OVER (PARTITION BY ... ORDER BY ...)). The expansion generates:
//! 1. A CTE that aggregates the inner metric(s) by ALL queried dimensions
//! 2. An outer SELECT that applies the window function over the CTE results
//!
//! PARTITION BY EXCLUDING is resolved at expansion time: the partition columns
//! are all queried dimensions MINUS the excluded dimensions.
```

The main function signature:
```rust
pub(super) fn expand_window_metrics(
    view_name: &str,
    def: &SemanticViewDefinition,
    resolved_dims: &[&crate::model::Dimension],
    resolved_mets: &[&Metric],
    resolved_exprs: &HashMap<String, String>,
    dim_scoped_aliases: &[Option<String>],
) -> Result<String, ExpandError>
```

Implementation logic:
a. **Validate required dimensions:** For each window metric, collect all dims from `window_spec.excluding_dims` and `window_spec.order_by[].expr`. Check each is present in `resolved_dims` (case-insensitive). If missing, return `Err(WindowMetricRequiredDimension { ... })`.

b. **Collect inner metrics needed for CTE:** For each window metric, resolve `window_spec.inner_metric` to its expression via `resolved_exprs`. The CTE must include these as aggregated columns.

c. **Build CTE `__sv_agg`:** Same pattern as semi_additive.rs but simpler:
   - SELECT: all queried dimensions (with their expressions and aliases), plus each distinct inner metric expression (aggregated, with alias)
   - FROM: base table + LEFT JOINs (same as regular expand)
   - GROUP BY: 1, 2, ... N (all dimension columns)

d. **Build outer SELECT:**
   - All queried dimension columns (referencing CTE aliases: `"dim_name"`)
   - Each window metric: `{window_function}({inner_metric_alias}{, extra_args}) OVER (PARTITION BY {computed_partition_cols} ORDER BY {order_entries} {frame_clause}) AS "{metric_name}"`
   - Computed partition cols = all queried dim names MINUS excluding_dims (case-insensitive set difference)
   - ORDER BY entries: dim/expr references as quoted CTE column aliases, with ASC/DESC and NULLS FIRST/LAST
   - Frame clause: raw string from WindowSpec.frame_clause, appended if present
   - FROM `__sv_agg`
   - NO outer GROUP BY

e. Include unit tests within the module (following semi_additive.rs pattern).

3. In src/expand/mod.rs:
   - Add `mod window;` declaration
   - No additional pub exports needed (window.rs is called from sql_gen.rs via super::window::expand_window_metrics)

4. In src/expand/sql_gen.rs expand() function:
   - AFTER the semi-additive dispatch (line ~334), BEFORE the regular SELECT generation:
   - Add window metric check and dispatch:

```rust
    // Phase 48: Check if any resolved metric is a window function metric.
    let has_window = resolved_mets.iter().any(|m| m.is_window());
    if has_window {
        // Window metrics cannot be mixed with aggregate metrics.
        let window_names: Vec<String> = resolved_mets
            .iter()
            .filter(|m| m.is_window())
            .map(|m| m.name.clone())
            .collect();
        let aggregate_names: Vec<String> = resolved_mets
            .iter()
            .filter(|m| !m.is_window())
            .map(|m| m.name.clone())
            .collect();
        if !aggregate_names.is_empty() {
            return Err(ExpandError::WindowAggregateMixing {
                view_name: view_name.to_string(),
                window_metrics: window_names,
                aggregate_metrics: aggregate_names,
            });
        }
        return super::window::expand_window_metrics(
            view_name,
            def,
            &resolved_dims,
            &resolved_mets,
            &resolved_exprs,
            &dim_scoped_aliases,
        );
    }
```

5. In src/expand/fan_trap.rs check_fan_traps():
   - Add a window metric skip right after the semi-additive skip (line ~67):
```rust
        // Phase 48: Skip fan trap check for window function metrics.
        // Window metrics operate on pre-aggregated CTE results, so fan-out
        // is handled by the inner aggregation step.
        if met.is_window() {
            continue;
        }
```

6. Write unit tests in src/expand/window.rs following the semi_additive.rs test module pattern, using test_helpers builders.
  </action>
  <verify>
    <automated>cargo test window && cargo test fan_trap && cargo test expand</automated>
  </verify>
  <acceptance_criteria>
    - File src/expand/window.rs exists and contains "pub(super) fn expand_window_metrics"
    - grep -c "mod window" src/expand/mod.rs returns 1
    - grep -c "WindowAggregateMixing" src/expand/types.rs returns at least 2 (variant + Display)
    - grep -c "WindowMetricRequiredDimension" src/expand/types.rs returns at least 2
    - grep -c "has_window" src/expand/sql_gen.rs returns at least 1
    - grep -c "is_window" src/expand/fan_trap.rs returns at least 1
    - grep "__sv_agg" src/expand/window.rs returns at least 1
    - grep "PARTITION BY" src/expand/window.rs returns at least 1
    - cargo test window exits 0 with at least 6 passing tests
    - cargo test fan_trap exits 0 with no failures
    - cargo test exits 0 (full suite, no regressions)
  </acceptance_criteria>
  <done>Window metric expansion generates CTE + outer SELECT with computed PARTITION BY. Mixing error blocks window + aggregate queries. Fan trap skips window metrics. Required dimension validation rejects queries missing EXCLUDING/ORDER BY dims. Unit tests pass for all paths.</done>
</task>

<task type="auto">
  <name>Task 2: SHOW DIMS required=TRUE and end-to-end integration tests</name>
  <files>src/ddl/show_dims_for_metric.rs, test/sql/phase48_window_metrics.test, test/sql/TEST_LIST</files>
  <read_first>
    - src/ddl/show_dims_for_metric.rs (full file -- ShowDimForMetricRow struct, bind() logic, func() constant FALSE)
    - src/expand/fan_trap.rs (collect_derived_metric_source_tables for reference)
    - test/sql/phase47_semi_additive.test (sqllogictest pattern for DDL + query + error tests)
    - test/sql/TEST_LIST (to add new test file entry)
  </read_first>
  <action>
1. In src/ddl/show_dims_for_metric.rs:
   - Add a `required: bool` field to ShowDimForMetricRow struct
   - In bind():
     - After finding the metric (`met`), check if `met.window_spec.is_some()`
     - If it is a window metric, collect the set of "required dimension names" from:
       - `window_spec.excluding_dims` (all of them -- the excluded dims must be queried)
       - `window_spec.order_by[].expr` (all ORDER BY dimensions must be queried)
       - Normalize to lowercase for comparison
     - When building ShowDimForMetricRow entries, set `required: true` if the dimension name (lowercased) is in the required set, `false` otherwise
     - For non-window metrics, `required` stays `false` (preserving existing behavior)
   - In func():
     - Replace the constant `false` loop with reading `row.required` from the bind data:
     ```rust
     for (i, row) in bind_data.rows.iter().enumerate() {
         // ... existing inserts ...
         req_slice[i] = row.required;
     }
     ```

2. Create test/sql/phase48_window_metrics.test with these test cases:

   **Setup:** Create tables (sales + dates) with test data, then CREATE SEMANTIC VIEW with:
   - A base aggregate metric: `s.total_qty AS SUM(s.quantity)`
   - A window metric: `s.avg_7_days AS AVG(total_qty) OVER (PARTITION BY EXCLUDING d.date ORDER BY d.date)`
   - A window metric with LAG: `s.qty_30_ago AS LAG(total_qty, 30) OVER (PARTITION BY EXCLUDING d.date ORDER BY d.date)`
   - Dimensions: d.date, d.year, s.store

   **Test cases (using sqllogictest statement/query syntax):**
   a. DDL: CREATE SEMANTIC VIEW succeeds (statement ok)
   b. Query window metric only: `SELECT * FROM semantic_view('v', dimensions := ['store', 'date', 'year'], metrics := ['avg_7_days'])` -- verify result columns and row count (query with type specifiers)
   c. Query with LAG window metric: verify LAG produces NULL for first row
   d. Mixed window + aggregate error: `SELECT * FROM semantic_view('v', dimensions := ['store', 'date'], metrics := ['total_qty', 'avg_7_days'])` -- statement error with "cannot mix window function metrics"
   e. Missing required dimension error: `SELECT * FROM semantic_view('v', dimensions := ['store'], metrics := ['avg_7_days'])` -- statement error with "requires dimension 'date'"
   f. SHOW SEMANTIC DIMENSIONS FOR METRIC: verify required=TRUE for date dimension, required=FALSE for others
   g. GET_DDL round-trip: `SELECT get_ddl('SEMANTIC_VIEW', 'v')` -- verify OVER clause appears in output
   h. DESCRIBE: verify WINDOW_SPEC property appears
   i. Fan trap: Create a view where window metric crosses many-to-one boundary -- no error (fan trap skipped)

3. Add `phase48_window_metrics` to test/sql/TEST_LIST (alphabetical order among phase entries).

4. Run `just build && just test-sql` to verify integration tests pass.
  </action>
  <verify>
    <automated>just test-all</automated>
  </verify>
  <acceptance_criteria>
    - grep -c "required: bool" src/ddl/show_dims_for_metric.rs returns 1 (in ShowDimForMetricRow struct)
    - grep -c "window_spec" src/ddl/show_dims_for_metric.rs returns at least 2
    - grep "row.required" src/ddl/show_dims_for_metric.rs shows the required field being read in func()
    - File test/sql/phase48_window_metrics.test exists
    - grep -c "statement ok" test/sql/phase48_window_metrics.test returns at least 2
    - grep -c "statement error" test/sql/phase48_window_metrics.test returns at least 2
    - grep -c "PARTITION BY EXCLUDING" test/sql/phase48_window_metrics.test returns at least 1
    - grep "phase48_window_metrics" test/sql/TEST_LIST returns 1 line
    - just test-all exits 0
  </acceptance_criteria>
  <done>SHOW SEMANTIC DIMENSIONS FOR METRIC returns required=TRUE for EXCLUDING/ORDER BY dims of window metrics. End-to-end sqllogictest covers DDL, query expansion, LAG, mixing error, required dim error, SHOW DIMS required, GET_DDL round-trip, DESCRIBE, and fan trap skip. Full quality gate passes (cargo test + sqllogictest + DuckLake CI).</done>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| Query request -> expand() | User-supplied dimension/metric names processed by expansion engine |
| CTE SQL generation | Generated SQL executed by DuckDB engine |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-48-04 | T (Tampering) | expand/window.rs CTE generation | mitigate | All identifiers passed through quote_ident(); frame clause from parsed DDL (not user query input); DuckDB validates generated SQL |
| T-48-05 | D (Denial of Service) | expand/window.rs set operations | accept | PARTITION BY EXCLUDING set difference is O(n*m) where n=queried dims, m=excluded dims; both are small (< 100 in practice) |
| T-48-06 | I (Information Disclosure) | SHOW DIMS required column | accept | required=TRUE reveals which dims are in EXCLUDING/ORDER BY but this info is already in GET_DDL output and DDL source |
</threat_model>

<verification>
After all tasks:
- `just test-all` passes (cargo test + sqllogictest + DuckLake CI)
- Window metric DDL -> storage -> query -> results works end-to-end
- Window + aggregate mixing blocked with clear error
- Fan trap detection does not trigger on window metrics
- SHOW DIMS FOR METRIC shows required=TRUE for window partition/order dims
- GET_DDL round-trips window metric OVER clause
</verification>

<success_criteria>
- CTE-based expansion generates correct SQL for window metrics (aggregate CTE + outer window SELECT)
- PARTITION BY EXCLUDING resolved to computed partition columns at query time
- Multiple window metrics in same query each have correct PARTITION BY
- Window + aggregate mixing produces blocking error with metric names listed
- Missing required dimensions produce clear error with dimension name and reason
- Fan trap check skips window metrics
- SHOW DIMS FOR METRIC required=TRUE for EXCLUDING and ORDER BY dims
- Full quality gate passes: `just test-all` green
</success_criteria>

<output>
After completion, create `.planning/phases/48-window-function-metrics/48-02-SUMMARY.md`
</output>
