---
phase: 02-storage-and-ddl
plan: "01"
type: execute
wave: 1
depends_on: []
files_modified:
  - Cargo.toml
  - src/model.rs
  - src/catalog.rs
autonomous: true
requirements:
  - DDL-01
  - DDL-02
  - DDL-03
  - DDL-04
  - DDL-05

must_haves:
  truths:
    - "SemanticViewDefinition Rust struct deserializes valid JSON and rejects JSON missing required fields"
    - "init_catalog creates semantic_layer schema and _definitions table idempotently on an in-memory DuckDB connection"
    - "catalog_insert writes a row and updates the HashMap; a second call with the same name returns an error"
    - "catalog_delete removes the row and drops the key from the HashMap; calling it for a non-existent name returns an error"
  artifacts:
    - path: "src/model.rs"
      provides: "SemanticViewDefinition struct with serde deserialization"
      contains: "struct SemanticViewDefinition"
    - path: "src/catalog.rs"
      provides: "CatalogState type alias, init_catalog, catalog_insert, catalog_delete helpers"
      exports: ["CatalogState", "init_catalog", "catalog_insert", "catalog_delete"]
    - path: "Cargo.toml"
      provides: "serde_json dep + vscalar feature added to duckdb"
      contains: "serde_json"
  key_links:
    - from: "src/catalog.rs"
      to: "semantic_layer._definitions"
      via: "execute_batch CREATE TABLE IF NOT EXISTS"
      pattern: "semantic_layer\\._definitions"
    - from: "src/catalog.rs"
      to: "src/model.rs"
      via: "SemanticViewDefinition::validate"
      pattern: "SemanticViewDefinition"
---

<objective>
Lay the data model and catalog persistence foundation for Phase 2's DDL functions.

Purpose: The scalar and table functions in plans 02-02 and 02-03 need two shared building blocks: (1) a validated Rust representation of a semantic view definition, and (2) helpers that initialize the DuckDB catalog table and read/write/delete definition rows while keeping the in-memory HashMap in sync.

Output: `src/model.rs` (SemanticViewDefinition + validation), `src/catalog.rs` (CatalogState, init_catalog, catalog_insert, catalog_delete), and updated `Cargo.toml` (serde_json + vscalar feature).
</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/02-storage-and-ddl/02-CONTEXT.md
@.planning/phases/02-storage-and-ddl/02-RESEARCH.md
@.planning/phases/01-scaffold/01-01-SUMMARY.md
@Cargo.toml
@src/lib.rs
</context>

<tasks>

<task type="auto">
  <name>Task 1: Add serde_json dependency and vscalar feature to Cargo.toml; create SemanticViewDefinition model</name>
  <files>Cargo.toml, src/model.rs</files>
  <action>
Update `Cargo.toml`:
- Change the duckdb dependency to: `duckdb = { version = "=1.4.4", features = ["loadable-extension", "vscalar"] }`
- Add: `serde = { version = "1", features = ["derive"] }`
- Add: `serde_json = "1"`

Create `src/model.rs` with the following:

```rust
use serde::{Deserialize, Serialize};

/// A named SQL column expression used as a dimension.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Dimension {
    pub name: String,
    pub expr: String,
}

/// A named aggregation expression used as a metric.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Metric {
    pub name: String,
    pub expr: String,
}

/// A JOIN relationship between the base table and another source table.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Join {
    pub table: String,
    pub on: String,
}

/// Top-level definition of a semantic view.
///
/// Stored as JSON in `semantic_layer._definitions`.
/// Required fields: `base_table`, `dimensions`, `metrics`.
/// Optional fields: `filters` (defaults to []), `joins` (defaults to []).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SemanticViewDefinition {
    pub base_table: String,
    pub dimensions: Vec<Dimension>,
    pub metrics: Vec<Metric>,
    #[serde(default)]
    pub filters: Vec<String>,
    #[serde(default)]
    pub joins: Vec<Join>,
}

impl SemanticViewDefinition {
    /// Parse and validate a JSON string, returning a typed definition.
    ///
    /// Returns an error if the JSON is invalid or missing required fields.
    /// The `name` parameter is used only in the error message for context.
    pub fn from_json(name: &str, json: &str) -> Result<Self, String> {
        serde_json::from_str(json).map_err(|e| {
            format!(
                "invalid definition for semantic view '{}': {}",
                name, e
            )
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn valid_definition_roundtrips() {
        let json = r#"{
            "base_table": "orders",
            "dimensions": [{"name": "region", "expr": "region"}],
            "metrics": [{"name": "revenue", "expr": "sum(amount)"}]
        }"#;
        let def = SemanticViewDefinition::from_json("orders", json).unwrap();
        assert_eq!(def.base_table, "orders");
        assert_eq!(def.dimensions.len(), 1);
        assert_eq!(def.metrics.len(), 1);
        assert!(def.filters.is_empty());
        assert!(def.joins.is_empty());
    }

    #[test]
    fn missing_base_table_is_error() {
        let json = r#"{"dimensions": [], "metrics": []}"#;
        assert!(SemanticViewDefinition::from_json("test", json).is_err());
    }

    #[test]
    fn invalid_json_is_error() {
        assert!(SemanticViewDefinition::from_json("test", "{not json}").is_err());
    }

    #[test]
    fn unknown_fields_are_rejected() {
        let json = r#"{"base_table": "t", "dimensions": [], "metrics": [], "extra": 1}"#;
        assert!(SemanticViewDefinition::from_json("test", json).is_err());
    }

    #[test]
    fn optional_fields_default_to_empty() {
        let json = r#"{"base_table": "t", "dimensions": [], "metrics": []}"#;
        let def = SemanticViewDefinition::from_json("test", json).unwrap();
        assert!(def.filters.is_empty());
        assert!(def.joins.is_empty());
    }
}
```

Add `pub mod model;` to `src/lib.rs` (just the module declaration; do not change the entry point function yet).

Run `cargo test` to verify the model tests pass. Run `cargo clippy -- -D warnings` to verify zero lint violations.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test model:: -- --nocapture 2>&1 | tail -20</automated>
    <manual>Confirm all 5 model tests pass: valid_definition_roundtrips, missing_base_table_is_error, invalid_json_is_error, unknown_fields_are_rejected, optional_fields_default_to_empty</manual>
  </verify>
  <done>
    - `cargo test model::` exits 0 with 5 tests passing
    - `cargo clippy -- -D warnings` exits 0 with zero violations
    - `Cargo.toml` contains `serde_json = "1"` and `"vscalar"` in duckdb features
    - `src/model.rs` contains `pub struct SemanticViewDefinition` and `fn from_json`
  </done>
</task>

<task type="auto">
  <name>Task 2: Implement catalog.rs — CatalogState, init_catalog, catalog_insert, catalog_delete</name>
  <files>src/catalog.rs, src/lib.rs</files>
  <action>
Create `src/catalog.rs`.

The catalog helpers manage two layers: the `semantic_layer._definitions` DuckDB table (authoritative) and the in-memory `HashMap` (load-time cache). The write-catalog-first pattern is used: write to the table first, update the HashMap only on success.

**CRITICAL pitfall:** A `duckdb::Connection` is not `Send + Sync`, so it cannot be stored in `CatalogState`. Instead, `init_catalog` receives a `&Connection` for the initial load, and `catalog_insert`/`catalog_delete` also receive a `&Connection`. These are called from the extension entrypoint (not from within VScalar invoke). See RESEARCH.md Open Question 1 — to avoid the second-connection concurrency issue, catalog writes will be routed through the entrypoint connection passed to the extension, not from within VScalar::invoke directly. This means the DDL functions will need a separate approach for catalog writes — see Task 1 of plan 02-02 for the scalar function design that handles this.

For now, implement the catalog module assuming the Connection is provided by the caller:

```rust
use std::{
    collections::HashMap,
    sync::{Arc, RwLock},
};

use duckdb::{Connection, Result};

use crate::model::SemanticViewDefinition;

/// Shared in-memory cache of semantic view definitions.
/// Key: view name. Value: raw JSON string of the definition.
pub type CatalogState = Arc<RwLock<HashMap<String, String>>>;

/// Create the `semantic_layer` schema and `_definitions` table if they do not exist,
/// then load all existing rows into a new [`CatalogState`].
///
/// This function is called once at extension load time. It is idempotent: safe to call
/// on every extension load regardless of whether the catalog already exists.
pub fn init_catalog(con: &Connection) -> Result<CatalogState> {
    con.execute_batch(
        "CREATE SCHEMA IF NOT EXISTS semantic_layer;
         CREATE TABLE IF NOT EXISTS semantic_layer._definitions (
             name       VARCHAR PRIMARY KEY,
             definition VARCHAR
         );",
    )?;

    let mut map = HashMap::new();
    let mut stmt =
        con.prepare("SELECT name, definition FROM semantic_layer._definitions")?;
    let rows = stmt.query_map([], |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
    })?;
    for row in rows {
        let (name, def) = row?;
        map.insert(name, def);
    }

    Ok(Arc::new(RwLock::new(map)))
}

/// Write a new semantic view definition to the catalog and update the in-memory cache.
///
/// Returns an error if:
/// - A view with `name` already exists (catalog PRIMARY KEY violation)
/// - The catalog write fails for any other reason
///
/// The HashMap is updated only on successful catalog write.
pub fn catalog_insert(
    con: &Connection,
    state: &CatalogState,
    name: &str,
    json: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    // Validate JSON before writing — fail fast, nothing written on invalid input
    SemanticViewDefinition::from_json(name, json)
        .map_err(|e| Box::<dyn std::error::Error>::from(e))?;

    // Check for duplicate before catalog write for a cleaner error message
    {
        let guard = state.read().unwrap();
        if guard.contains_key(name) {
            return Err(format!(
                "semantic view '{}' already exists; call drop_semantic_view first",
                name
            )
            .into());
        }
    }

    // Write to catalog first — error propagates via ? without touching HashMap
    con.execute(
        "INSERT INTO semantic_layer._definitions (name, definition) VALUES (?, ?)",
        duckdb::params![name, json],
    )?;

    // Update HashMap only on successful catalog write
    state.write().unwrap().insert(name.to_string(), json.to_string());
    Ok(())
}

/// Remove a semantic view definition from the catalog and the in-memory cache.
///
/// Returns an error if no view with `name` exists.
pub fn catalog_delete(
    con: &Connection,
    state: &CatalogState,
    name: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    {
        let guard = state.read().unwrap();
        if !guard.contains_key(name) {
            return Err(
                format!("semantic view '{}' does not exist", name).into()
            );
        }
    }

    // Write to catalog first
    let rows_affected = con.execute(
        "DELETE FROM semantic_layer._definitions WHERE name = ?",
        duckdb::params![name],
    )?;
    if rows_affected == 0 {
        // Defensive: catalog and HashMap were out of sync
        return Err(
            format!("semantic view '{}' does not exist", name).into()
        );
    }

    // Update HashMap only on successful catalog delete
    state.write().unwrap().remove(name);
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use duckdb::Connection;

    fn in_memory_con() -> Connection {
        Connection::open_in_memory().expect("in-memory DuckDB")
    }

    #[test]
    fn init_catalog_creates_schema_and_table() {
        let con = in_memory_con();
        let state = init_catalog(&con).unwrap();
        assert!(state.read().unwrap().is_empty());
        // Idempotent: second call must not error
        let state2 = init_catalog(&con).unwrap();
        assert!(state2.read().unwrap().is_empty());
    }

    #[test]
    fn insert_and_retrieve() {
        let con = in_memory_con();
        let state = init_catalog(&con).unwrap();
        let json = r#"{"base_table":"orders","dimensions":[],"metrics":[]}"#;
        catalog_insert(&con, &state, "orders", json).unwrap();
        let guard = state.read().unwrap();
        assert_eq!(guard.get("orders").map(String::as_str), Some(json));
    }

    #[test]
    fn duplicate_insert_is_error() {
        let con = in_memory_con();
        let state = init_catalog(&con).unwrap();
        let json = r#"{"base_table":"orders","dimensions":[],"metrics":[]}"#;
        catalog_insert(&con, &state, "orders", json).unwrap();
        let result = catalog_insert(&con, &state, "orders", json);
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(msg.contains("already exists"), "unexpected: {msg}");
    }

    #[test]
    fn delete_removes_from_hashmap_and_catalog() {
        let con = in_memory_con();
        let state = init_catalog(&con).unwrap();
        let json = r#"{"base_table":"orders","dimensions":[],"metrics":[]}"#;
        catalog_insert(&con, &state, "orders", json).unwrap();
        catalog_delete(&con, &state, "orders").unwrap();
        assert!(!state.read().unwrap().contains_key("orders"));
    }

    #[test]
    fn delete_nonexistent_is_error() {
        let con = in_memory_con();
        let state = init_catalog(&con).unwrap();
        let result = catalog_delete(&con, &state, "nonexistent");
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(msg.contains("does not exist"), "unexpected: {msg}");
    }

    #[test]
    fn init_catalog_loads_existing_rows() {
        let con = in_memory_con();
        // First load: insert a row
        let state = init_catalog(&con).unwrap();
        let json = r#"{"base_table":"orders","dimensions":[],"metrics":[]}"#;
        catalog_insert(&con, &state, "orders", json).unwrap();
        // Second load: simulates restart — loads from catalog
        let state2 = init_catalog(&con).unwrap();
        assert!(state2.read().unwrap().contains_key("orders"));
    }
}
```

Add `pub mod catalog;` to `src/lib.rs`.

Run `cargo test catalog::` and `cargo clippy -- -D warnings`.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test catalog:: -- --nocapture 2>&1 | tail -25</automated>
    <manual>Confirm all 6 catalog tests pass: init_catalog_creates_schema_and_table, insert_and_retrieve, duplicate_insert_is_error, delete_removes_from_hashmap_and_catalog, delete_nonexistent_is_error, init_catalog_loads_existing_rows</manual>
  </verify>
  <done>
    - `cargo test catalog::` exits 0 with 6 tests passing
    - `cargo clippy -- -D warnings` exits 0 with zero violations
    - `src/catalog.rs` exports: `CatalogState`, `init_catalog`, `catalog_insert`, `catalog_delete`
    - `init_catalog` uses `CREATE SCHEMA IF NOT EXISTS semantic_layer` and `CREATE TABLE IF NOT EXISTS semantic_layer._definitions`
    - Write-catalog-first pattern used in both `catalog_insert` and `catalog_delete`
  </done>
</task>

</tasks>

<verification>
After both tasks:
- `cargo test` exits 0 — all model and catalog unit tests pass (minimum 11 tests)
- `cargo clippy -- -D warnings` exits 0 — zero lint violations
- `cargo build` exits 0 — extension still compiles as cdylib
- `Cargo.toml` contains `serde_json = "1"`, `serde = { version = "1", features = ["derive"] }`, and duckdb features include `"vscalar"`
</verification>

<success_criteria>
- `src/model.rs` exists with `SemanticViewDefinition`, `Dimension`, `Metric`, `Join`, and `SemanticViewDefinition::from_json`
- `src/catalog.rs` exists with `CatalogState` type alias and `init_catalog`, `catalog_insert`, `catalog_delete` functions
- All unit tests pass; zero clippy violations; extension still builds
- Duplicate define returns "already exists" error; missing delete/describe returns "does not exist" error
</success_criteria>

<output>
After completion, create `.planning/phases/02-storage-and-ddl/02-01-SUMMARY.md`
</output>
