---
phase: 12-explain-typed-output
plan: 03
type: execute
wave: 2
depends_on:
  - 12-01
files_modified:
  - src/query/table_function.rs
autonomous: true
requirements:
  - OUT-01

must_haves:
  truths:
    - "`bind()` reads `column_types_inferred` from the stored SemanticViewDefinition and declares typed output columns (not all VARCHAR)"
    - "A BIGINT metric is declared as BIGINT in bind() and data is written as i64 bytes in func()"
    - "A DATE time dimension is declared as DATE in bind() and data is written as i32 (days-since-epoch) bytes in func()"
    - "A DOUBLE metric is declared as DOUBLE in bind() and data is written as f64 bytes in func()"
    - "When `column_types_inferred` is empty or a column maps to a complex/invalid type, the column falls back to VARCHAR (existing behavior)"
    - "NULL values are handled correctly for all typed columns (validity mask)"
  artifacts:
    - path: "src/query/table_function.rs"
      provides: "Typed bind() + typed func() dispatch for BIGINT/DATE/DOUBLE/INTEGER/VARCHAR"
      contains: "column_types_inferred"
  key_links:
    - from: "src/query/table_function.rs"
      to: "src/model.rs"
      via: "SemanticViewDefinition.column_types_inferred read in bind()"
      pattern: "column_types_inferred"
---

<objective>
Update `SemanticViewVTab::bind()` to read stored `column_types_inferred` from the catalog JSON and declare typed output columns. Update `func()` to dispatch typed vector writing for BIGINT, INTEGER, DOUBLE, DATE, and VARCHAR columns.

Purpose: This delivers OUT-01 — `semantic_view` returns correctly typed columns instead of all-VARCHAR.
Output: Updated `src/query/table_function.rs` with typed bind + func. No changes to test files (Plan 04 handles those).
</objective>

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

<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/12-explain-typed-output/12-RESEARCH.md
@.planning/phases/12-explain-typed-output/12-01-SUMMARY.md

<interfaces>
<!-- Key patterns from RESEARCH.md for typed output. No codebase re-exploration needed. -->

**duckdb_type → LogicalTypeHandle mapping:**
```rust
use duckdb::core::{LogicalTypeHandle, LogicalTypeId};
use libduckdb_sys as ffi;

fn type_from_duckdb_type(t: u32) -> LogicalTypeHandle {
    // Complex/invalid types fall back to VARCHAR
    match t {
        ffi::DUCKDB_TYPE_DUCKDB_TYPE_INVALID
        | ffi::DUCKDB_TYPE_DUCKDB_TYPE_LIST
        | ffi::DUCKDB_TYPE_DUCKDB_TYPE_STRUCT
        | ffi::DUCKDB_TYPE_DUCKDB_TYPE_MAP
        | ffi::DUCKDB_TYPE_DUCKDB_TYPE_ENUM => {
            LogicalTypeHandle::from(LogicalTypeId::Varchar)
        }
        // DECIMAL falls back to DOUBLE (avoids width/scale complexity)
        ffi::DUCKDB_TYPE_DUCKDB_TYPE_DECIMAL => {
            LogicalTypeHandle::from(LogicalTypeId::Double)
        }
        _ => LogicalTypeHandle::from(LogicalTypeId::from(t)),
    }
}
```

**FlatVector typed write (from RESEARCH.md):**
```rust
// BIGINT (i64):
let dst = out_vec.as_mut_slice_with_len::<i64>(n_rows);
dst.copy_from_slice(values_i64);

// DATE and INTEGER (i32 — DATE stored as days since epoch in DuckDB):
let dst = out_vec.as_mut_slice_with_len::<i32>(n_rows);
dst.copy_from_slice(values_i32);

// DOUBLE (f64):
let dst = out_vec.as_mut_slice_with_len::<f64>(n_rows);
dst.copy_from_slice(values_f64);

// NULL: out_vec.set_null(row_idx);
```

**Reading raw typed data from a result chunk vector:**
```rust
let src_vec = ffi::duckdb_data_chunk_get_vector(chunk, col_idx as ffi::idx_t);
let data_ptr = ffi::duckdb_vector_get_data(src_vec).cast::<T>();
let values = std::slice::from_raw_parts(data_ptr, row_count);
```

**Validity mask (NULL check) for source vector:**
```rust
let validity = ffi::duckdb_vector_get_validity(src_vec);
let is_null = if !validity.is_null() {
    let entry = *validity.add(row_idx / 64);
    (entry & (1u64 << (row_idx % 64))) == 0
} else {
    false
};
```

**SemanticViewBindData — current struct (Plan 03 extends it):**
```rust
pub struct SemanticViewBindData {
    expanded_sql: String,
    column_names: Vec<String>,
    // Plan 03: ADD column_type_ids: Vec<LogicalTypeId>
    //   — parallel to column_names, used by func() to dispatch typed write
}
```

**bind() current flow:**
```
1. Extract view_name + dimensions + metrics
2. Validate at least one requested
3. Look up catalog JSON
4. Parse SemanticViewDefinition from JSON
5. expand() → expanded_sql
6. infer_schema_or_default() → column_names (names only, types discarded)
7. bind.add_result_column(name, Varchar) for each column
```

**bind() Plan 03 changes:**
```
After step 4 (def now has column_type_names + column_types_inferred):
  4a. Build type_map: HashMap<String, u32> from def.column_type_names zip def.column_types_inferred
  4b. Build requested_column_names from dimensions + metrics (same order as expand() output)
  4c. If type_map is non-empty:
        For each requested column name, look up type in type_map (case-insensitive).
        Missing → 0u32 (INVALID → VARCHAR fallback).
        column_names = requested_column_names (no LIMIT 0 needed at query time)
      Else (type_map empty = in-memory DB or DDL inference failed):
        Call try_infer_schema(state.conn, expanded_sql + " LIMIT 0") → (names, types)
        Use returned names and types directly. Fallback: VARCHAR for all.
  Replace step 7: bind.add_result_column(name, type_from_duckdb_type_u32(type_id)) for each column
  Store column_names + column_type_ids in SemanticViewBindData for func()
```

Note: `infer_schema_or_default()` was used to get column names. Plan 03 still needs the names.

**Name-based lookup approach (matches CONTEXT.md locked decision — DDL-time as primary):**

Plan 02 stores BOTH `column_type_names: Vec<String>` and `column_types_inferred: Vec<u32>` in the
catalog JSON. These are parallel vecs covering ALL dims+metrics at DDL time.

At bind time, only a subset of dims+metrics may be requested — positional indexing won't work.
Instead, build a name→type HashMap:

```rust
// Build lookup map from DDL-time stored data (O(n) once at bind time):
let type_map: HashMap<String, u32> = def.column_type_names
    .iter()
    .zip(def.column_types_inferred.iter())
    .map(|(name, &t)| (name.to_lowercase(), t))
    .collect();

// For each requested column, look up by name (case-insensitive):
let column_type_ids: Vec<u32> = requested_column_names.iter()
    .map(|name| *type_map.get(&name.to_lowercase()).unwrap_or(&0u32))
    .collect();
// 0u32 = DUCKDB_TYPE_INVALID → VARCHAR fallback
```

If `type_map` is empty (in-memory DB, or DDL-time inference failed), fall back to bind-time
`try_infer_schema` on `state.conn` (safe in bind context, per RESEARCH.md). This is the fallback
path only — not the primary path.

Column names: derive from the existing `infer_schema_or_default` call OR from the requested
dimensions/metrics list (same order as expand() output). The name-based map lookup handles
type resolution; `infer_schema_or_default` is only needed for names when the type_map is empty
and we fall back to bind-time inference.
</interfaces>
</context>

<tasks>

<task type="auto">
  <name>Task 1: Update bind() to declare typed output columns, update func() for typed writes</name>
  <files>src/query/table_function.rs</files>
  <action>
    **Step 1: Add `column_type_ids: Vec<u32>` to `SemanticViewBindData`**

    Extend `SemanticViewBindData`:
    ```rust
    pub struct SemanticViewBindData {
        expanded_sql: String,
        column_names: Vec<String>,
        /// DuckDB type enums (as u32) for each output column, parallel to column_names.
        /// DUCKDB_TYPE_INVALID (0) signals VARCHAR fallback for that column.
        column_type_ids: Vec<u32>,
    }
    ```

    **Step 2: Add `type_from_duckdb_type_u32` helper function**

    Add a private helper (not pub) near the top of the VTab section:
    ```rust
    /// Convert a raw `ffi::duckdb_type` (u32) to a `LogicalTypeHandle`.
    /// Complex types (LIST, STRUCT, MAP, ENUM) and DECIMAL fall back to VARCHAR/DOUBLE.
    fn type_from_duckdb_type_u32(t: u32) -> LogicalTypeHandle {
        // DUCKDB_TYPE_INVALID, LIST, STRUCT, MAP, ENUM → VARCHAR fallback
        // DUCKDB_TYPE_DECIMAL → DOUBLE (avoids width/scale complexity)
        const INVALID: u32 = 0;  // DUCKDB_TYPE_DUCKDB_TYPE_INVALID
        const DECIMAL: u32 = 19; // DUCKDB_TYPE_DUCKDB_TYPE_DECIMAL
        const ENUM: u32 = 21;    // DUCKDB_TYPE_DUCKDB_TYPE_ENUM
        const LIST: u32 = 22;    // DUCKDB_TYPE_DUCKDB_TYPE_LIST
        const STRUCT: u32 = 23;  // DUCKDB_TYPE_DUCKDB_TYPE_STRUCT
        const MAP: u32 = 25;     // DUCKDB_TYPE_DUCKDB_TYPE_MAP
        match t {
            INVALID | ENUM | LIST | STRUCT | MAP => {
                LogicalTypeHandle::from(LogicalTypeId::Varchar)
            }
            DECIMAL => LogicalTypeHandle::from(LogicalTypeId::Double),
            _ => LogicalTypeHandle::from(LogicalTypeId::from(t)),
        }
    }
    ```

    Verify the constants match libduckdb-sys by cross-referencing with `ffi::DUCKDB_TYPE_DUCKDB_TYPE_*` constants in the codebase. Use the ffi constants directly in the match if they are available as const values in scope.

    **Step 3: Update `bind()` to use name-based type lookup from DDL-time stored data**

    Replace the `infer_schema_or_default` call in `bind()` with the following logic:

    ```rust
    // Build name→type map from DDL-time stored data (primary path, no per-query LIMIT 0).
    let type_map: std::collections::HashMap<String, u32> = def.column_type_names
        .iter()
        .zip(def.column_types_inferred.iter())
        .map(|(name, &t)| (name.to_lowercase(), t))
        .collect();

    let (column_names, column_type_ids) = if !type_map.is_empty() {
        // Primary path: derive column names from requested dims+metrics (expand() order),
        // look up each name in the DDL-time type map. No LIMIT 0 at query time.
        let mut names = Vec::new();
        let mut type_ids = Vec::new();
        for dim_name in &dimensions {
            let canonical = def.dimensions.iter()
                .find(|d| d.name.eq_ignore_ascii_case(dim_name))
                .map_or_else(|| dim_name.clone(), |d| d.name.clone());
            let t = *type_map.get(&canonical.to_lowercase()).unwrap_or(&0u32);
            names.push(canonical);
            type_ids.push(t);
        }
        for met_name in &metrics {
            let canonical = def.metrics.iter()
                .find(|m| m.name.eq_ignore_ascii_case(met_name))
                .map_or_else(|| met_name.clone(), |m| m.name.clone());
            let t = *type_map.get(&canonical.to_lowercase()).unwrap_or(&0u32);
            names.push(canonical);
            type_ids.push(t);
        }
        (names, type_ids)
    } else {
        // Fallback path: in-memory DB or DDL inference failed.
        // Run try_infer_schema on the per-query expanded SQL (bind-time, safe on state.conn).
        let limit0_sql = format!("{expanded_sql} LIMIT 0");
        if let Some((names, types)) = unsafe { try_infer_schema(state.conn, &limit0_sql) } {
            let type_ids = types.iter().map(|t| *t as u32).collect();
            (names, type_ids)
        } else {
            // Full fallback: derive names from definition, all VARCHAR.
            let mut names = Vec::new();
            for dim_name in &dimensions {
                let canonical = def.dimensions.iter()
                    .find(|d| d.name.eq_ignore_ascii_case(dim_name))
                    .map_or_else(|| dim_name.clone(), |d| d.name.clone());
                names.push(canonical);
            }
            for met_name in &metrics {
                let canonical = def.metrics.iter()
                    .find(|m| m.name.eq_ignore_ascii_case(met_name))
                    .map_or_else(|| met_name.clone(), |m| m.name.clone());
                names.push(canonical);
            }
            let type_ids = vec![0u32; names.len()];
            (names, type_ids)
        }
    };

    for (name, type_id) in column_names.iter().zip(column_type_ids.iter()) {
        bind.add_result_column(name, type_from_duckdb_type_u32(*type_id));
    }

    Ok(SemanticViewBindData {
        expanded_sql,
        column_names,
        column_type_ids,
    })
    ```

    Remove `infer_schema_or_default` (it is replaced by the logic above). Keep `try_infer_schema` as `pub(crate)` — still used in the fallback path.

    **Step 5: Update `func()` for typed writes**

    Replace the current all-VARCHAR approach in `func()`. The key changes:
    - Remove the `build_varchar_cast_sql` wrapping for typed columns
    - Execute the expanded SQL directly (no VARCHAR cast wrapper)
    - For each column, dispatch based on `bind_data.column_type_ids[col_idx]`

    The new func() structure:
    ```rust
    fn func(func: &TableFunctionInfo<Self>, output: &mut DataChunkHandle) -> Result<(), Box<dyn std::error::Error>> {
        let init_data = func.get_init_data();
        if init_data.done.swap(true, Ordering::Relaxed) {
            output.set_len(0);
            return Ok(());
        }

        let bind_data = func.get_bind_data();
        let state = unsafe { &*func.get_extra_info::<QueryState>() };

        // Execute the expanded SQL directly (no VARCHAR cast wrapper needed for typed columns).
        let mut result = unsafe {
            execute_sql_raw(state.conn, &bind_data.expanded_sql)
                .map_err(|e| QueryError::SqlExecution {
                    expanded_sql: bind_data.expanded_sql.clone(),
                    duckdb_error: e,
                })?
        };

        let col_count = unsafe { ffi::duckdb_column_count(&mut result) } as usize;
        let chunk_count = unsafe { ffi::duckdb_result_chunk_count(result) } as usize;

        // Collect total row count first for sizing output.
        let mut total_rows = 0usize;
        // Collect all row data per column, typed.
        // Use a column-major layout for efficient typed copy.
        let mut col_data: Vec<Vec<TypedValue>> = vec![Vec::new(); col_count];

        for chunk_idx in 0..chunk_count {
            let chunk = unsafe { ffi::duckdb_result_get_chunk(result, chunk_idx as ffi::idx_t) };
            if chunk.is_null() { continue; }
            let row_count = unsafe { ffi::duckdb_data_chunk_get_size(chunk) } as usize;
            total_rows += row_count;

            for col_idx in 0..col_count.min(bind_data.column_type_ids.len()) {
                let type_id = bind_data.column_type_ids[col_idx];
                // Read this column's values for this chunk and append to col_data[col_idx]
                unsafe {
                    read_typed_column(chunk, col_idx, row_count, type_id, &mut col_data[col_idx]);
                }
            }
            unsafe { ffi::duckdb_destroy_data_chunk(&mut { chunk }); }
        }

        // Write col_data to output.
        if total_rows > 0 {
            for col_idx in 0..col_count.min(bind_data.column_names.len()) {
                let type_id = if col_idx < bind_data.column_type_ids.len() {
                    bind_data.column_type_ids[col_idx]
                } else { 0 };
                write_typed_column(output, col_idx, &col_data[col_idx], total_rows, type_id);
            }
        }
        output.set_len(total_rows);
        unsafe { ffi::duckdb_destroy_result(&mut result); }
        Ok(())
    }
    ```

    **Implement `TypedValue`, `read_typed_column`, and `write_typed_column` helpers:**

    ```rust
    enum TypedValue {
        Null,
        I64(i64),   // BIGINT, HUGEINT (truncated), SMALLINT (upcast), UTINYINT etc
        I32(i32),   // INTEGER, DATE (days-since-epoch)
        F64(f64),   // DOUBLE, FLOAT (upcast)
        Str(String), // VARCHAR, all other types (fallback)
    }
    ```

    `read_typed_column`: Based on `type_id`, read from the raw vector data using `duckdb_vector_get_data`. For each row, check validity mask (NULL → TypedValue::Null). Then:
    - BIGINT (20), UBIGINT (22), HUGEINT (15 — read as i64, truncating upper half): read as i64
    - INTEGER (4), UINTEGER (21), SMALLINT (3), TINYINT (2), USMALLINT (19?), UTINYINT (18?), DATE (13): read as i32
    - DOUBLE (8), FLOAT (7): read as f64 (FLOAT upcast to f64 via safe from_bits or simple cast)
    - VARCHAR (17) and everything else: use existing `read_varchar_from_vector` → TypedValue::Str

    IMPORTANT: Use the actual `ffi::DUCKDB_TYPE_*` constant values from `libduckdb_sys`. Do not hardcode magic numbers — use the named constants. Look up the constants in the project's vendored/generated bindings:
    ```rust
    ffi::DUCKDB_TYPE_DUCKDB_TYPE_BIGINT     // i64
    ffi::DUCKDB_TYPE_DUCKDB_TYPE_INTEGER    // i32
    ffi::DUCKDB_TYPE_DUCKDB_TYPE_DATE       // i32 (days since epoch)
    ffi::DUCKDB_TYPE_DUCKDB_TYPE_DOUBLE     // f64
    ffi::DUCKDB_TYPE_DUCKDB_TYPE_FLOAT      // f32 → cast to f64
    ffi::DUCKDB_TYPE_DUCKDB_TYPE_VARCHAR    // String
    ```

    `write_typed_column`: Using `bind_data.column_type_ids[col_idx]`, call the appropriate write path:
    - I64 values → `out_vec.as_mut_slice_with_len::<i64>(n)` + `copy_from_slice`, NULLs via `set_null`
    - I32 values → `out_vec.as_mut_slice_with_len::<i32>(n)` + `copy_from_slice`, NULLs via `set_null`
    - F64 values → `out_vec.as_mut_slice_with_len::<f64>(n)` + `copy_from_slice`, NULLs via `set_null`
    - Str values → existing `out_vec.insert(row_idx, s.as_str())` path with empty string for NULL

    Keep `build_varchar_cast_sql` in the file — it may still be useful for the explain path — but do not call it from the main `func()`.

    Run `cargo test` to verify all tests pass.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test 2>&1 | tail -20</automated>
  </verify>
  <done>
    `cargo test` passes. `SemanticViewBindData` has `column_type_ids: Vec<u32>`. `bind()` calls `infer_schema_with_types` and declares typed columns. `func()` dispatches typed vector writes. `TypedValue`, `read_typed_column`, `write_typed_column` helpers are implemented.
  </done>
</task>

</tasks>

<verification>
`cargo test` passes. Key grep checks:
- `grep "column_type_ids" src/query/table_function.rs` — multiple hits
- `grep "as_mut_slice_with_len" src/query/table_function.rs` — hits on i64/i32/f64 writes
- `grep "TypedValue" src/query/table_function.rs` — hits on enum definition and usage
</verification>

<success_criteria>
`cargo test` passes. `semantic_view` bind declares typed columns (BIGINT, DATE, DOUBLE, INTEGER as appropriate) based on DDL-time inference results. `func()` writes typed vector data without VARCHAR wrapping for numeric and date types. VARCHAR fallback retained for complex/unknown types.
</success_criteria>

<output>
After completion, create `.planning/phases/12-explain-typed-output/12-03-SUMMARY.md`
</output>
