---
phase: 25-sql-body-parser
plan: 03
type: execute
wave: 3
depends_on:
  - 25-02
files_modified:
  - src/parse.rs
  - src/ddl/define.rs
  - src/lib.rs
autonomous: true
requirements:
  - DDL-01
  - DDL-07

must_haves:
  truths:
    - "CREATE SEMANTIC VIEW name AS TABLES (...) DIMENSIONS (...) METRICS (...) is accepted by validate_and_rewrite"
    - "Rewritten SQL routes through create_semantic_view_from_json VTab (not old body syntax)"
    - "CREATE OR REPLACE and CREATE IF NOT EXISTS also route through _from_json variants"
    - "DROP, DROP IF EXISTS, DESCRIBE, SHOW are unchanged (still rewrite to existing functions)"
    - "parse_proptest AS-body round-trip test now passes (validate_and_rewrite returns Ok(Some(_)))"
  artifacts:
    - path: "src/parse.rs"
      provides: "AS dispatch in validate_create_body; rewrite_ddl_keyword_body producing JSON-in-SQL"
      contains: "create_semantic_view_from_json"
    - path: "src/ddl/define.rs"
      provides: "DefineFromJsonVTab implementing create_semantic_view_from_json"
      contains: "DefineFromJsonVTab"
    - path: "src/lib.rs"
      provides: "Registration of 3 _from_json VTab functions in init_extension"
      contains: "create_semantic_view_from_json"
  key_links:
    - from: "src/parse.rs::validate_create_body"
      to: "src/body_parser.rs::parse_keyword_body"
      via: "AS detection route; parse result JSON-encoded and embedded in rewritten SQL"
      pattern: "parse_keyword_body"
    - from: "cpp/src/shim.cpp::sv_ddl_bind"
      to: "src/ddl/define.rs::DefineFromJsonVTab"
      via: "rewritten SQL: SELECT * FROM create_semantic_view_from_json('name', 'json')"
      pattern: "create_semantic_view_from_json"
    - from: "src/ddl/define.rs::DefineFromJsonVTab"
      to: "src/model.rs::SemanticViewDefinition::from_json"
      via: "deserializes JSON param to SemanticViewDefinition, then same persist/catalog logic"
      pattern: "from_json"
---

<objective>
Wire the body parser into the DDL pipeline: add AS dispatch in `parse.rs` and add JSON-accepting VTab functions in `define.rs`.

Purpose: Plan 02 built the parser; this plan connects it to DuckDB. `validate_create_body` must detect the `AS` keyword path and route to `parse_keyword_body`, then serialize the result to JSON for the rewritten SQL. Three new VTab functions (`create_semantic_view_from_json`, `create_or_replace_semantic_view_from_json`, `create_semantic_view_if_not_exists_from_json`) accept `(name VARCHAR, json VARCHAR)` and delegate to the existing persist/catalog logic.

Output: Full end-to-end pipeline from `CREATE SEMANTIC VIEW name AS ...` DDL to stored definition.
</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/25-sql-body-parser/25-CONTEXT.md
@.planning/phases/25-sql-body-parser/25-RESEARCH.md
@.planning/phases/25-sql-body-parser/25-02-SUMMARY.md
@src/parse.rs
@src/ddl/define.rs
@src/lib.rs
@src/body_parser.rs

<interfaces>
<!-- Functions to call and types to use in this plan. -->

From src/body_parser.rs (Plan 02 output):
```rust
pub fn parse_keyword_body(text: &str, base_offset: usize) -> Result<KeywordBody, ParseError>;

pub struct KeywordBody {
    pub tables: Vec<TableRef>,
    pub relationships: Vec<Join>,
    pub dimensions: Vec<Dimension>,
    pub metrics: Vec<Metric>,
}
// parse_keyword_body returns Ok(KeywordBody) with a .into_definition(name) method
// OR the caller constructs SemanticViewDefinition from the KeywordBody fields.
// Check body_parser.rs output from Plan 02 for the exact API.
```

From src/parse.rs (functions to modify):
```rust
// Current validate_create_body detects '(' only. Must also detect 'AS'.
fn validate_create_body(
    query: &str,
    trimmed_no_semi: &str,
    trim_offset: usize,
    plen: usize,
) -> Result<Option<String>, ParseError>;

// Helper already available:
pub struct ParseError { pub message: String, pub position: Option<usize> }
fn prefix_len(kind: DdlKind) -> usize;
fn function_name(kind: DdlKind) -> &'static str;
```

From src/ddl/define.rs (existing patterns to follow for new VTabs):
```rust
// DefineState is shared across all 3 CREATE variants:
pub struct DefineState {
    pub catalog: CatalogState,
    pub persist_conn: Option<ffi::duckdb_connection>,
    pub or_replace: bool,
    pub if_not_exists: bool,
}

// persist_define is the shared persistence function — reuse it:
fn persist_define(conn: ffi::duckdb_connection, name: &str, json: &str) -> Result<(), String>;

// catalog_insert and catalog_upsert from crate::catalog:
use crate::catalog::{catalog_insert, catalog_upsert};
```

From src/model.rs:
```rust
impl SemanticViewDefinition {
    pub fn from_json(name: &str, json: &str) -> Result<Self, String>;
}
```
</interfaces>
</context>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Add AS dispatch in parse.rs validate_create_body and rewrite_ddl_keyword_body</name>
  <files>src/parse.rs</files>
  <behavior>
    - Test: validate_and_rewrite("CREATE SEMANTIC VIEW v AS TABLES (t AS orders PRIMARY KEY (id)) DIMENSIONS (t.r AS r) METRICS (t.m AS SUM(1))") returns Ok(Some(sql)) where sql starts with "SELECT * FROM create_semantic_view_from_json("
    - Test: validate_and_rewrite("CREATE OR REPLACE SEMANTIC VIEW v AS TABLES ...") routes to create_or_replace_semantic_view_from_json
    - Test: validate_and_rewrite("CREATE SEMANTIC VIEW IF NOT EXISTS v AS TABLES ...") routes to create_semantic_view_if_not_exists_from_json
    - Test: validate_and_rewrite("CREATE SEMANTIC VIEW v AS TABLSE (...) ...") returns Err with "did you mean TABLES" position pointing at "TABLSE"
    - Test: validate_and_rewrite("CREATE SEMANTIC VIEW v AS TABLES (...) DIMENSIONS (...) METRICS (...)") rewritten SQL contains the view name quoted
    - Test: validate_and_rewrite("CREATE SEMANTIC VIEW v (tables := [])") still works (old paren path unchanged)
    - Test: validate_and_rewrite("DROP SEMANTIC VIEW v") still returns Ok(Some("SELECT * FROM drop_semantic_view('v')")) (unchanged)
  </behavior>
  <action>
Modify `src/parse.rs`:

**Step 1: Import body_parser.**
At the top of `parse.rs`, add:
```rust
use crate::body_parser::parse_keyword_body;
```

**Step 2: Add `rewrite_ddl_keyword_body` private function.**
Add a new function that handles the AS-body rewrite path:
```rust
/// Rewrite an AS-body CREATE DDL statement to a JSON-parameterized function call.
///
/// Called when validate_create_body detects the `AS` keyword path.
/// Parses the keyword body via body_parser, serializes to JSON, and embeds in
/// a `SELECT * FROM create_semantic_view_from_json('name', 'json')` call.
fn rewrite_ddl_keyword_body(
    query: &str,
    kind: DdlKind,
    name: &str,
    body_text: &str,   // text starting at "AS" (inclusive)
    body_offset: usize, // byte offset of body_text[0] in original query
) -> Result<Option<String>, ParseError> {
    // 1. Call parse_keyword_body (body_text starts at "AS"; pass body_offset)
    let keyword_body = parse_keyword_body(body_text, body_offset)?;

    // 2. Construct SemanticViewDefinition from KeywordBody
    //    base_table = first table's physical table name (backward compat)
    let base_table = keyword_body.tables.first()
        .map(|t| t.table.clone())
        .unwrap_or_default();

    let def = crate::model::SemanticViewDefinition {
        base_table,
        tables: keyword_body.tables,
        dimensions: keyword_body.dimensions,
        metrics: keyword_body.metrics,
        joins: keyword_body.relationships,
        filters: vec![],
        facts: vec![],
        column_type_names: vec![],
        column_types_inferred: vec![],
    };

    // 3. Serialize to JSON
    let json = serde_json::to_string(&def).map_err(|e| ParseError {
        message: format!("Failed to serialize definition: {e}"),
        position: None,
    })?;

    // 4. SQL-escape single quotes in name and JSON
    let safe_name = name.replace('\'', "''");
    let safe_json = json.replace('\'', "''");

    // 5. Pick the correct _from_json function name based on DDL kind
    let fn_name = match kind {
        DdlKind::Create => "create_semantic_view_from_json",
        DdlKind::CreateOrReplace => "create_or_replace_semantic_view_from_json",
        DdlKind::CreateIfNotExists => "create_semantic_view_if_not_exists_from_json",
        _ => unreachable!("rewrite_ddl_keyword_body only called for CREATE forms"),
    };

    Ok(Some(format!("SELECT * FROM {fn_name}('{safe_name}', '{safe_json}')")))
}
```

**Step 3: Modify `validate_create_body`.**
In the existing `validate_create_body` function, after extracting the view name (`name` and `name_end`), add AS detection before the existing `(` check:

```rust
// Detect AS keyword path (new) vs ( path (old).
let after_name = &after_prefix[name_end..].trim_start();
if after_name.eq_ignore_ascii_case("as")
    || after_name.to_ascii_lowercase().starts_with("as ")
    || after_name.to_ascii_lowercase().starts_with("as\t")
    || after_name.to_ascii_lowercase().starts_with("as\n")
{
    // AS keyword body path
    // body_text starts at the position of "as" in the original query
    let as_offset_in_after_prefix = name_end + (after_prefix[name_end..].len() - after_name.len());
    let body_offset = trim_offset + plen + (trimmed_no_semi.len() - plen - after_prefix.len()) + as_offset_in_after_prefix;
    let body_text = &after_name[..]; // everything from "AS" onwards
    return rewrite_ddl_keyword_body(query, kind, name, body_text, body_offset);
}
// else: fall through to existing '(' detection
```

Note: Be careful with offset arithmetic. The key invariant is that `query[body_offset..]` starts at the "AS" keyword in the original query string. Thread trim_offset through correctly.

**Position arithmetic guidance:**
- `trim_offset` = number of leading spaces in `query` (already computed in `validate_and_rewrite`)
- `plen` = prefix length (e.g., "create semantic view".len())
- After prefix, `after_prefix` is the trimmed remainder
- `name_end` = end of name in `after_prefix`
- The gap between `trimmed_no_semi[plen..]` and `after_prefix` accounts for whitespace stripping

Use `query.find("AS")` as a sanity check but NOT for position computation (would find wrong "AS" in complex queries). Compute via offsets.

**Step 4: Add parse tests for the new path.**
Add tests inside the existing `#[cfg(test)] mod tests` block of `parse.rs`:
```rust
mod phase25_parse_tests {
    use super::*;

    #[test]
    fn as_body_create_rewrites_to_from_json() {
        let query = "CREATE SEMANTIC VIEW v AS TABLES (t AS orders PRIMARY KEY (id)) DIMENSIONS (t.r AS r) METRICS (t.m AS SUM(1))";
        let result = validate_and_rewrite(query).unwrap().unwrap();
        assert!(result.starts_with("SELECT * FROM create_semantic_view_from_json("), "Got: {result}");
        assert!(result.contains("'v'"), "Must contain view name: {result}");
    }

    #[test]
    fn as_body_create_or_replace_rewrites_to_from_json() {
        let query = "CREATE OR REPLACE SEMANTIC VIEW v AS TABLES (t AS orders PRIMARY KEY (id)) DIMENSIONS (t.r AS r) METRICS (t.m AS SUM(1))";
        let result = validate_and_rewrite(query).unwrap().unwrap();
        assert!(result.starts_with("SELECT * FROM create_or_replace_semantic_view_from_json("), "Got: {result}");
    }

    #[test]
    fn as_body_create_if_not_exists_rewrites_to_from_json() {
        let query = "CREATE SEMANTIC VIEW IF NOT EXISTS v AS TABLES (t AS orders PRIMARY KEY (id)) DIMENSIONS (t.r AS r) METRICS (t.m AS SUM(1))";
        let result = validate_and_rewrite(query).unwrap().unwrap();
        assert!(result.starts_with("SELECT * FROM create_semantic_view_if_not_exists_from_json("), "Got: {result}");
    }

    #[test]
    fn old_paren_body_still_works() {
        let query = "CREATE SEMANTIC VIEW v (tables := [], dimensions := [])";
        let result = validate_and_rewrite(query);
        assert!(result.is_ok(), "Old paren path must still work: {result:?}");
    }

    #[test]
    fn drop_still_rewrites_unchanged() {
        let query = "DROP SEMANTIC VIEW v";
        let result = validate_and_rewrite(query).unwrap().unwrap();
        assert_eq!(result, "SELECT * FROM drop_semantic_view('v')");
    }
}
```

IMPORTANT: The old `( ... )` path in `validate_create_body` must remain fully functional. Only ADD the AS detection before the existing paren check — do NOT remove or modify any existing code paths.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test parse::phase25_parse_tests 2>&1 | tail -15</automated>
  </verify>
  <done>
    - `cargo test parse::phase25_parse_tests` shows all 5 tests green
    - `validate_and_rewrite` routes AS-body DDL to `create_semantic_view_from_json` variants
    - Old paren-body and name-only DDL forms still pass their existing tests
    - `cargo test` (full suite) passes with no regressions
  </done>
</task>

<task type="auto">
  <name>Task 2: Add DefineFromJsonVTab to define.rs and register in lib.rs</name>
  <files>src/ddl/define.rs, src/lib.rs</files>
  <action>
Add the three `_from_json` VTab implementations to `src/ddl/define.rs` and register them in `src/lib.rs`.

**In src/ddl/define.rs:**

Add `DefineFromJsonVTab` after the existing `DefineSemanticViewVTab` implementation (around line 237). This struct implements `VTab` with exactly 2 positional parameters (`name VARCHAR`, `json VARCHAR`) and no named parameters:

```rust
/// `create_semantic_view_from_json(name, json)` table function.
///
/// Accepts a pre-parsed JSON-serialized `SemanticViewDefinition` (produced by
/// the AS-body DDL rewriter in parse.rs). Deserializes and stores the definition
/// using the same catalog/persist logic as `DefineSemanticViewVTab`.
///
/// This is the execution target for `CREATE SEMANTIC VIEW name AS TABLES (...) ...`.
pub struct DefineFromJsonVTab;

impl VTab for DefineFromJsonVTab {
    type BindData = DefineBindData;
    type InitData = DefineInitData;

    fn bind(bind: &BindInfo) -> Result<Self::BindData, Box<dyn std::error::Error>> {
        bind.add_result_column("view_name", LogicalTypeHandle::from(LogicalTypeId::Varchar));

        let name = bind.get_parameter(0).to_string();
        let json = bind.get_parameter(1).to_string();

        // Deserialize the JSON into a SemanticViewDefinition.
        let mut def = crate::model::SemanticViewDefinition::from_json(&name, &json)
            .map_err(|e| Box::<dyn std::error::Error>::from(e))?;

        // Access the DefineState from extra_info.
        let state_ptr = bind.get_extra_info::<DefineState>();
        let state = unsafe { &*state_ptr };

        // DDL-time type inference (same as DefineSemanticViewVTab).
        if let Some(conn) = state.persist_conn {
            let req_all = crate::expand::QueryRequest {
                dimensions: def.dimensions.iter().map(|d| d.name.clone()).collect(),
                metrics: def.metrics.iter().map(|m| m.name.clone()).collect(),
            };
            if let Ok(expanded_for_inference) = crate::expand::expand(&name, &def, &req_all) {
                let limit0_sql = format!("{expanded_for_inference} LIMIT 0");
                if let Some((names, types)) =
                    unsafe { crate::query::table_function::try_infer_schema(conn, &limit0_sql) }
                {
                    def.column_type_names = names;
                    def.column_types_inferred = types
                        .iter()
                        .map(|t| crate::query::table_function::normalize_type_id(*t as u32))
                        .collect();
                }
            }
        }

        let json_out = serde_json::to_string(&def)
            .map_err(|e| Box::<dyn std::error::Error>::from(e.to_string()))?;

        // Persist to DuckDB table first (file-backed databases only).
        if let Some(conn) = state.persist_conn {
            persist_define(conn, &name, &json_out)
                .map_err(|e| Box::<dyn std::error::Error>::from(e))?;
        }

        // Update in-memory catalog.
        if state.or_replace {
            catalog_upsert(&state.catalog, &name, &json_out)?;
        } else if state.if_not_exists {
            match catalog_insert(&state.catalog, &name, &json_out) {
                Ok(()) => {}
                Err(e) if e.to_string().contains("already exists") => {}
                Err(e) => return Err(e),
            }
        } else {
            catalog_insert(&state.catalog, &name, &json_out)?;
        }

        Ok(DefineBindData { name })
    }

    fn init(_: &InitInfo) -> Result<Self::InitData, Box<dyn std::error::Error>> {
        Ok(DefineInitData { done: AtomicBool::new(false) })
    }

    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 name_vec = output.flat_vector(0);
        name_vec.insert(0, bind_data.name.as_str());
        output.set_len(1);
        Ok(())
    }

    fn parameters() -> Option<Vec<LogicalTypeHandle>> {
        // Both name and json are positional VARCHAR parameters.
        Some(vec![
            LogicalTypeHandle::from(LogicalTypeId::Varchar), // name
            LogicalTypeHandle::from(LogicalTypeId::Varchar), // json
        ])
    }

    fn named_parameters() -> Option<Vec<(String, LogicalTypeHandle)>> {
        None // No named parameters — both are positional
    }
}
```

`DefineFromJsonVTab` reuses `DefineBindData`, `DefineInitData`, and `DefineState` from the same module — no new types needed.

**In src/lib.rs:**

In the `use crate::ddl::define` import block (inside the `extension` module), add `DefineFromJsonVTab`:
```rust
use crate::ddl::define::{DefineFromJsonVTab, DefineSemanticViewVTab, DefineState};
```

Then register the three `_from_json` variants in `init_extension`, immediately after the existing three `create_semantic_view*` registrations:

```rust
// Register create_semantic_view_from_json(name, json) -- target for AS-body DDL
con.register_table_function_with_extra_info::<DefineFromJsonVTab, _>(
    "create_semantic_view_from_json",
    &define_state,
)?;

// Register create_or_replace_semantic_view_from_json(name, json)
con.register_table_function_with_extra_info::<DefineFromJsonVTab, _>(
    "create_or_replace_semantic_view_from_json",
    &define_or_replace_state,
)?;

// Register create_semantic_view_if_not_exists_from_json(name, json)
con.register_table_function_with_extra_info::<DefineFromJsonVTab, _>(
    "create_semantic_view_if_not_exists_from_json",
    &define_if_not_exists_state,
)?;
```

`define_state`, `define_or_replace_state`, and `define_if_not_exists_state` are already constructed above these lines — just reuse them as-is.

IMPORTANT: Do NOT modify any existing VTab registrations. Do NOT change `DefineSemanticViewVTab`. The old function-call DDL path must continue to work alongside the new AS-body path.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test 2>&1 | tail -10</automated>
  </verify>
  <done>
    - `DefineFromJsonVTab` compiles without errors
    - Three `_from_json` functions registered in `init_extension`
    - `cargo test` passes (all unit tests green, no regressions)
    - `just build` succeeds (extension binary built)
  </done>
</task>

</tasks>

<verification>
```bash
cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test 2>&1 | tail -5
cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && just build 2>&1 | tail -5
```
Full test suite passes. Extension builds. AS-body DDL pipeline is end-to-end connected.
</verification>

<success_criteria>
- `cargo test` passes with 0 failures (all existing + 5 new phase25_parse_tests green)
- `just build` succeeds (C++ + Rust compiles together)
- `validate_and_rewrite("CREATE SEMANTIC VIEW v AS TABLES (...) ...")` returns `Ok(Some("SELECT * FROM create_semantic_view_from_json('v', '...')"))`
- `DefineFromJsonVTab` registered under three function names in `init_extension`
- Old paren-body and name-only DDL forms unchanged and passing
</success_criteria>

<output>
After completion, create `.planning/phases/25-sql-body-parser/25-03-SUMMARY.md`
</output>
