---
phase: 39-metadata-storage
plan: 01
type: execute
wave: 1
depends_on: []
files_modified:
  - src/model.rs
  - src/body_parser.rs
  - src/parse.rs
  - src/ddl/define.rs
  - src/graph/test_helpers.rs
  - src/graph/relationship.rs
  - src/expand/sql_gen.rs
  - fuzz/fuzz_targets/fuzz_query_names.rs
  - tests/expand_proptest.rs
  - test/sql/phase39_metadata_storage.test
autonomous: true
requirements:
  - META-01
  - META-02
  - META-03
  - META-04
  - META-05

must_haves:
  truths:
    - "Newly created semantic views carry created_on, database_name, schema_name metadata"
    - "Fact structs carry an output_type field for downstream SHOW FACTS display"
    - "Pre-v0.5.5 stored JSON without new fields deserializes to None without error"
    - "All existing tests pass with zero behavior changes"
  artifacts:
    - path: "src/model.rs"
      provides: "SemanticViewDefinition with created_on, database_name, schema_name; Fact with output_type"
      contains: "pub created_on: Option<String>"
    - path: "src/ddl/define.rs"
      provides: "Metadata capture at bind time via execute_sql_raw on catalog_conn"
      contains: "strftime(now()"
    - path: "test/sql/phase39_metadata_storage.test"
      provides: "sqllogictest end-to-end verification of metadata storage"
      contains: "phase39"
  key_links:
    - from: "src/ddl/define.rs"
      to: "src/model.rs"
      via: "def.created_on = Some(...)"
      pattern: "def\\.created_on"
    - from: "src/body_parser.rs"
      to: "src/model.rs"
      via: "Fact { ... output_type: None }"
      pattern: "output_type: None"
---

<objective>
Add four new metadata fields to the semantic view model (created_on, database_name, schema_name on SemanticViewDefinition; output_type on Fact), capture them at define time, and ensure full backward compatibility with pre-v0.5.5 stored JSON.

Purpose: Downstream phases 40 (SHOW alignment) and 41 (DESCRIBE rewrite) need these metadata fields to display Snowflake-aligned column schemas.
Output: Updated model structs, metadata capture in define.rs, backward compat unit tests, end-to-end sqllogictest.
</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/39-metadata-storage/39-RESEARCH.md

<interfaces>
<!-- Key types and contracts the executor needs. -->

From src/model.rs (current Fact struct -- line 70):
```rust
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
pub struct Fact {
    pub name: String,
    pub expr: String,
    #[serde(default)]
    pub source_table: Option<String>,
}
```

From src/model.rs (current SemanticViewDefinition -- line 164):
```rust
pub struct SemanticViewDefinition {
    pub base_table: String,
    #[serde(default)]
    pub tables: Vec<TableRef>,
    pub dimensions: Vec<Dimension>,
    pub metrics: Vec<Metric>,
    #[serde(default)]
    pub joins: Vec<Join>,
    #[serde(default)]
    pub facts: Vec<Fact>,
    #[serde(default)]
    pub column_type_names: Vec<String>,
    #[serde(default)]
    pub column_types_inferred: Vec<u32>,
}
```

From src/ddl/define.rs (DefineState -- line 20):
```rust
pub struct DefineState {
    pub catalog: CatalogState,
    pub persist_conn: Option<ffi::duckdb_connection>,
    pub catalog_conn: ffi::duckdb_connection,
    pub or_replace: bool,
    pub if_not_exists: bool,
}
```

From src/query/table_function.rs:
```rust
pub(crate) unsafe fn execute_sql_raw(
    conn: ffi::duckdb_connection,
    sql: &str,
) -> Result<ffi::duckdb_result, String>;
```
</interfaces>
</context>

<tasks>

<task type="auto">
  <name>Task 1: Add model fields and fix all struct constructions</name>
  <files>
    src/model.rs,
    src/body_parser.rs,
    src/parse.rs,
    src/graph/test_helpers.rs,
    src/graph/relationship.rs,
    src/expand/sql_gen.rs,
    fuzz/fuzz_targets/fuzz_query_names.rs,
    tests/expand_proptest.rs
  </files>
  <read_first>
    src/model.rs,
    src/body_parser.rs,
    src/parse.rs,
    src/graph/test_helpers.rs,
    src/graph/relationship.rs,
    src/expand/sql_gen.rs,
    fuzz/fuzz_targets/fuzz_query_names.rs,
    tests/expand_proptest.rs
  </read_first>
  <action>
**Step 1 -- Add fields to model.rs structs:**

In `src/model.rs`, add three new fields to `SemanticViewDefinition` (after the `column_types_inferred` field, before the closing brace):

```rust
/// ISO 8601 timestamp of when this semantic view was created.
/// Captured at define time via DuckDB `now()`.
/// Old stored JSON without this field deserializes to None.
#[serde(default)]
pub created_on: Option<String>,

/// Database name from the connection context at define time.
/// Old stored JSON without this field deserializes to None.
#[serde(default)]
pub database_name: Option<String>,

/// Schema name from the connection context at define time.
/// Old stored JSON without this field deserializes to None.
#[serde(default)]
pub schema_name: Option<String>,
```

In `src/model.rs`, add one new field to the `Fact` struct (after `source_table`):

```rust
/// Optional output type for this fact, used by SHOW FACTS data_type column.
/// Populated at define time via type inference when possible.
/// Old stored JSON without this field deserializes to None.
#[serde(default)]
pub output_type: Option<String>,
```

**Step 2 -- Fix all explicit Fact struct constructions (21 occurrences across 4 files):**

Every `Fact { name, expr, source_table }` or `Fact { name, expr, source_table: None }` or `Fact { name, expr, source_table: Some(...) }` must add `output_type: None`.

Files and locations:
- `src/body_parser.rs` (line ~330): The `.map(|(alias, bare_name, expr)| Fact { ... })` -- add `output_type: None` after the `source_table: Some(alias)` line.
- `src/graph/test_helpers.rs` (line ~82): The `make_def_with_facts` helper `.map(|(name, expr, source)| Fact { ... })` -- add `output_type: None`.
- `src/expand/sql_gen.rs`: 18 occurrences in test code. Every `Fact { name: ..., expr: ..., source_table: ... }` gets `output_type: None` appended. Search for all `Fact {` patterns and update each one.

**Step 3 -- Fix all explicit SemanticViewDefinition struct constructions:**

Every `SemanticViewDefinition { ... }` that explicitly lists `column_type_names` and `column_types_inferred` must also add:
```rust
created_on: None,
database_name: None,
schema_name: None,
```

This applies to constructions in:
- `src/model.rs` tests: 3 explicit constructions (lines ~448, ~544, ~644). Two others use `..Default::default()` and are fine.
- `src/parse.rs` (line ~841): Add the three fields after `column_types_inferred: vec![]`.
- `src/graph/test_helpers.rs`: 5 explicit constructions in `make_def`, `make_def_with_facts`, `make_def_with_derived_metrics`, `make_def_with_named_joins`, and the return of `make_def_with_derived_metrics`.
- `src/graph/relationship.rs` tests: 2 explicit constructions (`legacy_empty_fk_columns_skips_validation` at line ~525, and `make_fk_ref_def` at line ~709).
- `src/expand/sql_gen.rs`: Many explicit constructions in test helpers (`orders_view`, `def_with_join_columns`, `pkfk_two_table_def`, `pkfk_three_table_def`, `qualified_ref_def`, and individual test functions). Approximately 15+ constructions.
- `fuzz/fuzz_targets/fuzz_query_names.rs`: Check for explicit constructions; those using `..Default::default()` are fine.
- `tests/expand_proptest.rs`: Check for explicit constructions; those using `..Default::default()` are fine.

**Step 4 -- Add unit tests to model.rs:**

Add a new test module `phase39_metadata_tests` in `src/model.rs` (after the existing `phase33_model_tests` module) with these tests:

```rust
mod phase39_metadata_tests {
    use super::*;

    #[test]
    fn created_on_roundtrip() {
        let mut def = SemanticViewDefinition {
            base_table: "orders".to_string(),
            dimensions: vec![],
            metrics: vec![],
            created_on: Some("2026-04-01T12:00:00Z".to_string()),
            database_name: Some("mydb".to_string()),
            schema_name: Some("main".to_string()),
            ..Default::default()
        };
        let json = serde_json::to_string(&def).unwrap();
        let rt = SemanticViewDefinition::from_json("orders", &json).unwrap();
        assert_eq!(rt.created_on.as_deref(), Some("2026-04-01T12:00:00Z"));
        assert_eq!(rt.database_name.as_deref(), Some("mydb"));
        assert_eq!(rt.schema_name.as_deref(), Some("main"));
    }

    #[test]
    fn old_json_without_metadata_fields_deserializes() {
        // Backward compat: pre-v0.5.5 JSON has no created_on/database_name/schema_name
        let json = r#"{"base_table":"orders","dimensions":[],"metrics":[]}"#;
        let def = SemanticViewDefinition::from_json("orders", json).unwrap();
        assert!(def.created_on.is_none(), "created_on should default to None");
        assert!(def.database_name.is_none(), "database_name should default to None");
        assert!(def.schema_name.is_none(), "schema_name should default to None");
    }

    #[test]
    fn fact_output_type_roundtrip() {
        let fact = Fact {
            name: "rev".to_string(),
            expr: "amount".to_string(),
            source_table: Some("orders".to_string()),
            output_type: Some("DECIMAL(10,2)".to_string()),
        };
        let json = serde_json::to_string(&fact).unwrap();
        let rt: Fact = serde_json::from_str(&json).unwrap();
        assert_eq!(rt.output_type.as_deref(), Some("DECIMAL(10,2)"));
    }

    #[test]
    fn old_fact_json_without_output_type_deserializes() {
        // Backward compat: pre-v0.5.5 Fact JSON has no output_type
        let json = r#"{"name":"rev","expr":"amount","source_table":"orders"}"#;
        let fact: Fact = serde_json::from_str(json).unwrap();
        assert!(fact.output_type.is_none(), "output_type should default to None");
    }
}
```

**IMPORTANT:** Do NOT include a `filters` field in any struct construction. The `filters` field was removed in a prior phase.

**Verify compilation:** Run `cargo test` after all changes to confirm zero compiler errors and all tests pass.
  </action>
  <verify>
    <automated>cargo test</automated>
  </verify>
  <acceptance_criteria>
    - src/model.rs contains `pub created_on: Option<String>`
    - src/model.rs contains `pub database_name: Option<String>`
    - src/model.rs contains `pub schema_name: Option<String>`
    - src/model.rs Fact struct contains `pub output_type: Option<String>`
    - src/model.rs contains `fn old_json_without_metadata_fields_deserializes`
    - src/model.rs contains `fn fact_output_type_roundtrip`
    - src/body_parser.rs Fact construction contains `output_type: None`
    - src/parse.rs SemanticViewDefinition construction contains `created_on: None`
    - src/graph/test_helpers.rs all Fact constructions contain `output_type: None`
    - src/graph/test_helpers.rs all SemanticViewDefinition constructions contain `created_on: None`
    - `cargo test` exits 0 with all tests passing
  </acceptance_criteria>
  <done>All four new Option String fields are added with serde(default), every explicit struct construction in the codebase compiles with the new fields, backward compatibility unit tests pass, and cargo test exits 0.</done>
</task>

<task type="auto">
  <name>Task 2: Capture metadata at define time and add sqllogictest</name>
  <files>
    src/ddl/define.rs,
    test/sql/phase39_metadata_storage.test
  </files>
  <read_first>
    src/ddl/define.rs,
    src/model.rs,
    src/query/table_function.rs,
    test/sql/phase34_1_show_commands.test
  </read_first>
  <action>
**Step 1 -- Add metadata capture to define.rs bind():**

In `DefineFromJsonVTab::bind()`, after the `validate_using_relationships` call (line ~236) and before the DDL-time type inference block (line ~240), insert metadata capture code:

```rust
// Capture metadata: timestamp, database_name, schema_name.
// Uses catalog_conn which is always available (file-backed AND in-memory).
let metadata_sql = "SELECT strftime(now(), '%Y-%m-%dT%H:%M:%SZ'), \
                    current_database(), current_schema()";
let metadata_result = unsafe {
    crate::query::table_function::execute_sql_raw(state.catalog_conn, metadata_sql)
};
if let Ok(mut result) = metadata_result {
    unsafe {
        let ts_ptr = ffi::duckdb_value_varchar(&mut result, 0, 0);
        if !ts_ptr.is_null() {
            def.created_on = Some(
                CStr::from_ptr(ts_ptr).to_string_lossy().into_owned()
            );
            ffi::duckdb_free(ts_ptr as *mut std::ffi::c_void);
        }
        let db_ptr = ffi::duckdb_value_varchar(&mut result, 1, 0);
        if !db_ptr.is_null() {
            def.database_name = Some(
                CStr::from_ptr(db_ptr).to_string_lossy().into_owned()
            );
            ffi::duckdb_free(db_ptr as *mut std::ffi::c_void);
        }
        let schema_ptr = ffi::duckdb_value_varchar(&mut result, 2, 0);
        if !schema_ptr.is_null() {
            def.schema_name = Some(
                CStr::from_ptr(schema_ptr).to_string_lossy().into_owned()
            );
            ffi::duckdb_free(schema_ptr as *mut std::ffi::c_void);
        }
        ffi::duckdb_destroy_result(&mut result);
    }
}
```

Note: `CStr` is already imported at the top of define.rs (`use std::ffi::CStr`).

**Step 2 -- Add fact output_type inference to define.rs bind():**

After the existing DDL-time type inference block (the `if let Some(conn) = state.persist_conn { ... }` block that runs `try_infer_schema`), add fact type inference:

```rust
// Fact type inference: use typeof(expr) to determine fact output types.
// Only attempt when catalog_conn is available and facts exist.
if !def.facts.is_empty() {
    // Build alias -> physical table name mapping
    let alias_to_table: std::collections::HashMap<String, String> = def
        .tables
        .iter()
        .map(|t| (t.alias.to_ascii_lowercase(), t.table.clone()))
        .collect();

    // Use catalog_conn (always available) or persist_conn for typeof queries.
    let type_conn = state.persist_conn.unwrap_or(state.catalog_conn);

    for fact in &mut def.facts {
        // Resolve the physical table name from the fact's source_table alias
        let table_name = fact
            .source_table
            .as_ref()
            .and_then(|alias| alias_to_table.get(&alias.to_ascii_lowercase()).cloned())
            .unwrap_or_else(|| def.base_table.clone());

        let sql = format!(
            "SELECT typeof({}) FROM \"{}\" LIMIT 1",
            fact.expr, table_name.replace('"', "\"\"")
        );
        if let Ok(mut result) = unsafe {
            crate::query::table_function::execute_sql_raw(type_conn, &sql)
        } {
            let row_count = unsafe { ffi::duckdb_row_count(&mut result) };
            if row_count > 0 {
                let val_ptr = unsafe { ffi::duckdb_value_varchar(&mut result, 0, 0) };
                if !val_ptr.is_null() {
                    let type_name = unsafe {
                        CStr::from_ptr(val_ptr).to_string_lossy().into_owned()
                    };
                    unsafe { ffi::duckdb_free(val_ptr as *mut std::ffi::c_void) };
                    fact.output_type = Some(type_name);
                }
            }
            unsafe { ffi::duckdb_destroy_result(&mut result) };
        }
        // If typeof() fails (expression references other facts, table empty, etc.),
        // output_type stays None -- graceful degradation for SHOW FACTS display.
    }
}
```

IMPORTANT: The fact type inference is best-effort. If `typeof()` fails (table is empty, expression references other facts in DAG order, etc.), `output_type` stays `None`. This is acceptable -- Phase 40 can show an empty string for `data_type` when `output_type` is None.

**Step 3 -- Create sqllogictest for end-to-end verification:**

Create `test/sql/phase39_metadata_storage.test`:

```
# Phase 39: Metadata Storage - verify created_on, database_name, schema_name, fact output_type

require semantic_views

# Setup: create table and semantic view
statement ok
CREATE TABLE p39_orders (id INTEGER PRIMARY KEY, amount DECIMAL(10,2), qty INTEGER);

statement ok
CREATE SEMANTIC VIEW p39_sv AS
  TABLES (
    o AS p39_orders PRIMARY KEY (id)
  )
  FACTS (
    o.unit_price AS o.amount / o.qty
  )
  DIMENSIONS (
    o.order_id AS o.id
  )
  METRICS (
    o.total_amount AS SUM(o.amount)
  )

# Test 1: Read back the stored JSON and verify metadata fields exist
# The definition is stored as JSON in the catalog. We query it via the catalog
# and verify the new fields are present.
# Since we can't directly inspect JSON in sqllogictest, we verify indirectly
# by creating the view successfully (no errors) and then querying it.

# Test 2: Query the view to ensure it still works with new fields
query IT
SELECT * FROM semantic_view('p39_sv', dimensions := ['order_id'], metrics := ['total_amount'])
----

# Test 3: Drop and recreate to verify roundtrip through catalog
statement ok
DROP SEMANTIC VIEW p39_sv

statement ok
CREATE SEMANTIC VIEW p39_sv AS
  TABLES (
    o AS p39_orders PRIMARY KEY (id)
  )
  FACTS (
    o.unit_price AS o.amount / o.qty
  )
  DIMENSIONS (
    o.order_id AS o.id
  )
  METRICS (
    o.total_amount AS SUM(o.amount)
  )

# Test 4: CREATE OR REPLACE preserves metadata capture
statement ok
CREATE OR REPLACE SEMANTIC VIEW p39_sv AS
  TABLES (
    o AS p39_orders PRIMARY KEY (id)
  )
  FACTS (
    o.unit_price AS o.amount / o.qty
  )
  DIMENSIONS (
    o.order_id AS o.id
  )
  METRICS (
    o.total_amount AS SUM(o.amount)
  )

# Test 5: Verify query still works after replace
query IT
SELECT * FROM semantic_view('p39_sv', dimensions := ['order_id'], metrics := ['total_amount'])
----

# Test 6: Cleanup
statement ok
DROP SEMANTIC VIEW p39_sv

statement ok
DROP TABLE p39_orders
```

**Step 4 -- Run full test suite:**

Run `just test-all` to confirm all Rust tests, sqllogictests, and DuckLake CI pass.
  </action>
  <verify>
    <automated>just test-all</automated>
  </verify>
  <acceptance_criteria>
    - src/ddl/define.rs contains `strftime(now(), '%Y-%m-%dT%H:%M:%SZ')`
    - src/ddl/define.rs contains `current_database()`
    - src/ddl/define.rs contains `current_schema()`
    - src/ddl/define.rs contains `def.created_on = Some(`
    - src/ddl/define.rs contains `def.database_name = Some(`
    - src/ddl/define.rs contains `def.schema_name = Some(`
    - src/ddl/define.rs contains `typeof(` for fact output_type inference
    - src/ddl/define.rs contains `fact.output_type = Some(`
    - test/sql/phase39_metadata_storage.test exists and contains `require semantic_views`
    - `just test-all` exits 0
  </acceptance_criteria>
  <done>Metadata (created_on, database_name, schema_name) is captured at define time via DuckDB SQL functions on catalog_conn. Fact output_type is inferred via typeof() on a best-effort basis. End-to-end sqllogictest confirms views with metadata fields create, query, drop, and recreate without error. just test-all passes.</done>
</task>

</tasks>

<verification>
1. `cargo test` passes -- all Rust unit tests, proptests, and doc tests green
2. `just test-all` passes -- includes sqllogictest and DuckLake CI
3. New model fields verified via unit tests: roundtrip serialization, backward compat with old JSON
4. No SHOW/DESCRIBE output changes (confirmed by existing sqllogictests passing unchanged)
</verification>

<success_criteria>
- SemanticViewDefinition has created_on, database_name, schema_name fields (Option String, serde default)
- Fact has output_type field (Option String, serde default)
- define.rs captures all three metadata fields at bind time via execute_sql_raw on catalog_conn
- define.rs attempts fact output_type inference via typeof() with graceful fallback to None
- Pre-v0.5.5 JSON without new fields deserializes cleanly (unit tested)
- just test-all exits 0 (all 482+ existing tests pass plus new tests)
</success_criteria>

<output>
After completion, create `.planning/phases/39-metadata-storage/39-01-SUMMARY.md`
</output>
