---
phase: 25-sql-body-parser
plan: 01
type: execute
wave: 1
depends_on: []
files_modified:
  - cpp/src/shim.cpp
  - src/body_parser.rs
  - test/sql/phase25_keyword_body.test
  - tests/parse_proptest.rs
autonomous: true
requirements:
  - DDL-01
  - DDL-07

must_haves:
  truths:
    - "C++ DDL buffer can handle 64 KB rewritten SQL without truncation"
    - "body_parser.rs module exists and compiles with unit test stubs for each clause parser"
    - "sqllogictest file exists with keyword-body CREATE and all 7 DDL verb skeletons"
    - "parse_proptest.rs has AS-body proptest block (failing tests stub waiting for implementation)"
  artifacts:
    - path: "cpp/src/shim.cpp"
      provides: "64 KB std::string buffer replacing char sql_buf[4096] in sv_ddl_bind and sv_parse_stub"
      contains: "std::string sql_str(65536"
    - path: "src/body_parser.rs"
      provides: "Module skeleton with parse_keyword_body stub and unit tests for clause parsers"
      contains: "pub fn parse_keyword_body"
    - path: "test/sql/phase25_keyword_body.test"
      provides: "sqllogictest file with keyword-body DDL tests"
      contains: "AS TABLES"
    - path: "tests/parse_proptest.rs"
      provides: "Proptest block for AS-body DDL round-trip and position invariants"
      contains: "as_body_keyword"
  key_links:
    - from: "cpp/src/shim.cpp::sv_ddl_bind"
      to: "sv_rewrite_ddl_rust"
      via: "sql_str.data() and sql_str.size()"
      pattern: "sql_str\\.data\\(\\)"
    - from: "src/body_parser.rs"
      to: "src/model.rs"
      via: "imports for TableRef, Join, Dimension, Metric"
      pattern: "use crate::model"
---

<objective>
Create the test infrastructure and fix the C++ buffer size limit before implementing the keyword body parser.

Purpose: The 4096-byte C++ stack buffer must be replaced with a 64 KB heap allocation before any integration test with a large keyword body can succeed. Unit test stubs in `body_parser.rs` establish the Nyquist contract (Wave 0 gap requirement) so subsequent implementation tasks have automated verification from the start. The sqllogictest file and proptest block define the acceptance criteria against which Wave 1 and Wave 2 work is validated.

Output: Fixed C++ buffer, compilable `body_parser.rs` skeleton with test stubs, sqllogictest test file, extended proptest block.
</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

<interfaces>
<!-- Key types and function signatures for this plan. Extracted from codebase. -->

From src/model.rs (after Phase 24 completes — these are the target types body_parser.rs will produce):
```rust
pub struct TableRef {
    pub alias: String,
    pub table: String,
    #[serde(default)]
    pub pk_columns: Vec<String>,
}

pub struct Join {
    pub table: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub on: String,
    #[serde(default)]
    pub from_cols: Vec<String>,
    #[serde(default)]
    pub join_columns: Vec<JoinColumn>,
    #[serde(default)]
    pub from_alias: String,
    #[serde(default)]
    pub fk_columns: Vec<String>,
    #[serde(default)]
    pub name: Option<String>,
}

pub struct Dimension {
    pub name: String,
    pub expr: String,
    #[serde(default)]
    pub source_table: Option<String>,
    #[serde(default)]
    pub output_type: Option<String>,
}

pub struct Metric {
    pub name: String,
    pub expr: String,
    #[serde(default)]
    pub source_table: Option<String>,
    #[serde(default)]
    pub output_type: Option<String>,
}

pub struct SemanticViewDefinition {
    pub base_table: String,
    #[serde(default)]
    pub tables: Vec<TableRef>,
    pub dimensions: Vec<Dimension>,
    pub metrics: Vec<Metric>,
    // ... other fields with serde(default)
}
```

From src/parse.rs (existing types body_parser.rs must use):
```rust
#[derive(Debug)]
pub struct ParseError {
    pub message: String,
    pub position: Option<usize>,
}
```

From cpp/src/shim.cpp (lines to fix):
```cpp
// Line 70 in sv_parse_stub — BEFORE:
char sql_buf[4096];
memset(sql_buf, 0, sizeof(sql_buf));
// Line 139 in sv_ddl_bind — BEFORE:
char sql_buf[4096];
memset(sql_buf, 0, sizeof(sql_buf));
```
</interfaces>
</context>

<tasks>

<task type="auto">
  <name>Task 1: Fix C++ DDL buffer — replace char sql_buf[4096] with std::string(65536)</name>
  <files>cpp/src/shim.cpp</files>
  <action>
In `cpp/src/shim.cpp`, fix both `char sql_buf[4096]` occurrences:

**sv_parse_stub (around line 70):**
Replace:
```cpp
char sql_buf[4096];
char error_buf[1024];
uint32_t position = UINT32_MAX;
memset(sql_buf, 0, sizeof(sql_buf));
memset(error_buf, 0, sizeof(error_buf));

uint8_t rc = sv_validate_ddl_rust(
    reinterpret_cast<const char *>(query.c_str()),
    query.size(),
    sql_buf, sizeof(sql_buf),
    error_buf, sizeof(error_buf),
    &position);
```
With:
```cpp
std::string sql_str(16384, '\0');  // 16 KB: validation path, not executed
char error_buf[1024];
uint32_t position = UINT32_MAX;
memset(error_buf, 0, sizeof(error_buf));

uint8_t rc = sv_validate_ddl_rust(
    reinterpret_cast<const char *>(query.c_str()),
    query.size(),
    sql_str.data(), sql_str.size(),
    error_buf, sizeof(error_buf),
    &position);
```

**sv_ddl_bind (around line 139):**
Replace:
```cpp
char sql_buf[4096];
char error_buf[1024];
memset(sql_buf, 0, sizeof(sql_buf));
memset(error_buf, 0, sizeof(error_buf));

uint8_t rc = sv_rewrite_ddl_rust(
    query.c_str(), query.size(),
    sql_buf, sizeof(sql_buf),
    error_buf, sizeof(error_buf));

if (rc != 0) {
    throw BinderException("Semantic view DDL failed: %s", error_buf);
}

// Step 2: Execute the rewritten SQL on the DDL connection
duckdb_result result;
if (duckdb_query(sv_ddl_conn, sql_buf, &result) != DuckDBSuccess) {
```
With:
```cpp
std::string sql_str(65536, '\0');  // 64 KB: execution path needs headroom for large views
char error_buf[1024];
memset(error_buf, 0, sizeof(error_buf));

uint8_t rc = sv_rewrite_ddl_rust(
    query.c_str(), query.size(),
    sql_str.data(), sql_str.size(),
    error_buf, sizeof(error_buf));

if (rc != 0) {
    throw BinderException("Semantic view DDL failed: %s", error_buf);
}

// Step 2: Execute the rewritten SQL on the DDL connection
duckdb_result result;
if (duckdb_query(sv_ddl_conn, sql_str.c_str(), &result) != DuckDBSuccess) {
```

Note: After the buffer fix, any reference to `sql_buf` in the execution path must be updated to `sql_str.c_str()`. Check for any other uses of `sql_buf` in sv_ddl_bind and update them.

Do NOT change any other logic — only the buffer allocation and the DDL connection query call. Do NOT add truncation detection logic (the 64 KB headroom covers all practical cases per RESEARCH.md buffer analysis).
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && just build 2>&1 | tail -5</automated>
  </verify>
  <done>
    - `sv_parse_stub` uses `std::string sql_str(16384, '\0')` instead of `char sql_buf[4096]`
    - `sv_ddl_bind` uses `std::string sql_str(65536, '\0')` instead of `char sql_buf[4096]`
    - `just build` succeeds (extension binary built without error)
    - No `char sql_buf[4096]` remains in shim.cpp
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 2: Create body_parser.rs skeleton with clause parser unit test stubs</name>
  <files>src/body_parser.rs, src/lib.rs</files>
  <behavior>
    - Test: parse_tables_clause parses "o AS orders PRIMARY KEY (o_orderkey)" -> TableRef { alias:"o", table:"orders", pk_columns:["o_orderkey"] }
    - Test: parse_tables_clause parses schema-qualified table "o AS main.orders PRIMARY KEY (o_orderkey)" -> TableRef { alias:"o", table:"main.orders", pk_columns:["o_orderkey"] }
    - Test: parse_tables_clause parses composite PK "(l_orderkey, l_linenumber)" -> pk_columns:["l_orderkey","l_linenumber"]
    - Test: parse_relationships_clause parses "order_to_customer AS o(customer_id) REFERENCES c" -> Join { from_alias:"o", fk_columns:["customer_id"], name:Some("order_to_customer"), table:"c", ... }
    - Test: parse_relationships_clause returns Ok(vec![]) for empty body ""
    - Test: parse_qualified_entries parses "o.revenue AS SUM(amount)" -> source_table:"o", name:"revenue", expr:"SUM(amount)"
    - Test: parse_qualified_entries parses expression with nested parens "o.disc_price AS SUM(l_extendedprice * (1 - l_discount))" verbatim
    - Test: parse_qualified_entries allows trailing comma (entry after last comma is discarded if empty)
    - Test: parse_keyword_body returns Err(ParseError) for input without TABLES keyword (stub: just panic or return Err)
  </behavior>
  <action>
Create `src/body_parser.rs` as a new file. This plan creates the skeleton with unit tests and minimal stub implementations that make tests FAIL correctly (red phase). The green phase is Plan 02.

Structure of body_parser.rs:
```rust
//! SQL keyword body parser for CREATE SEMANTIC VIEW.
//!
//! Parses: `AS TABLES (...) RELATIONSHIPS (...) DIMENSIONS (...) METRICS (...)`
//! into a `SemanticViewDefinition`.

use crate::model::{Dimension, Join, Metric, SemanticViewDefinition, TableRef};
use crate::parse::ParseError;

/// Result of parsing the keyword body (everything after "AS").
pub struct KeywordBody {
    pub tables: Vec<TableRef>,
    pub relationships: Vec<Join>,
    pub dimensions: Vec<Dimension>,
    pub metrics: Vec<Metric>,
}

/// Parse the keyword body after "AS" into structured clause data.
///
/// `text` is the full text after the "AS" keyword, trimmed.
/// `base_offset` is the byte offset of `text[0]` in the original query string.
pub fn parse_keyword_body(text: &str, base_offset: usize) -> Result<KeywordBody, ParseError> {
    todo!("Plan 02 implements this")
}

/// Parse the content inside TABLES (...).
pub(crate) fn parse_tables_clause(body: &str, base_offset: usize) -> Result<Vec<TableRef>, ParseError> {
    todo!("Plan 02 implements this")
}

/// Parse the content inside RELATIONSHIPS (...). Returns empty vec for empty body.
pub(crate) fn parse_relationships_clause(body: &str, base_offset: usize) -> Result<Vec<Join>, ParseError> {
    todo!("Plan 02 implements this")
}

/// Parse the content inside DIMENSIONS or METRICS (...).
/// Returns Vec<(source_alias, bare_name, expr)>.
pub(crate) fn parse_qualified_entries(body: &str, base_offset: usize) -> Result<Vec<(String, String, String)>, ParseError> {
    todo!("Plan 02 implements this")
}

/// Split `body` at depth-0 commas, respecting nested parens and single-quoted strings.
/// Returns Vec<(start_offset_in_body, trimmed_slice)>. Trailing empty entries discarded.
pub(crate) fn split_at_depth0_commas(body: &str) -> Vec<(usize, &str)> {
    // Implementation stub — returns empty until Plan 02.
    // Plan 02 replaces this with the depth-tracking algorithm from RESEARCH.md Pattern.
    vec![]
}

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

    // --- TABLES clause ---

    #[test]
    #[should_panic]  // Remove should_panic in Plan 02 when implementation is complete
    fn parse_tables_single_pk() {
        let result = parse_tables_clause("o AS orders PRIMARY KEY (o_orderkey)", 0).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].alias, "o");
        assert_eq!(result[0].table, "orders");
        assert_eq!(result[0].pk_columns, vec!["o_orderkey"]);
    }

    #[test]
    #[should_panic]
    fn parse_tables_schema_qualified() {
        let result = parse_tables_clause("o AS main.orders PRIMARY KEY (o_orderkey)", 0).unwrap();
        assert_eq!(result[0].table, "main.orders");
        assert_eq!(result[0].alias, "o");
    }

    #[test]
    #[should_panic]
    fn parse_tables_composite_pk() {
        let result = parse_tables_clause("l AS lineitem PRIMARY KEY (l_orderkey, l_linenumber)", 0).unwrap();
        assert_eq!(result[0].pk_columns, vec!["l_orderkey", "l_linenumber"]);
    }

    // --- RELATIONSHIPS clause ---

    #[test]
    fn parse_relationships_empty_body() {
        // Empty relationships body — must return Ok(vec![]) WITHOUT should_panic
        // This case works as a stub returning todo!() only if called...
        // Temporarily skip until Plan 02.
    }

    #[test]
    #[should_panic]
    fn parse_relationships_single_entry() {
        let result = parse_relationships_clause(
            "order_to_customer AS o(customer_id) REFERENCES c", 0
        ).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].name.as_deref(), Some("order_to_customer"));
        assert_eq!(result[0].from_alias, "o");
        assert_eq!(result[0].fk_columns, vec!["customer_id"]);
        assert_eq!(result[0].table, "c");
    }

    // --- DIMENSIONS / METRICS clause ---

    #[test]
    #[should_panic]
    fn parse_qualified_entries_simple() {
        let result = parse_qualified_entries("o.revenue AS SUM(amount)", 0).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].0, "o");          // source_alias
        assert_eq!(result[0].1, "revenue");    // bare_name
        assert_eq!(result[0].2, "SUM(amount)"); // expr
    }

    #[test]
    #[should_panic]
    fn parse_qualified_entries_nested_parens() {
        let result = parse_qualified_entries(
            "o.disc_price AS SUM(l_extendedprice * (1 - l_discount))", 0
        ).unwrap();
        assert_eq!(result[0].2, "SUM(l_extendedprice * (1 - l_discount))");
    }

    #[test]
    #[should_panic]
    fn parse_qualified_entries_trailing_comma() {
        let result = parse_qualified_entries("o.revenue AS SUM(amount),", 0).unwrap();
        assert_eq!(result.len(), 1, "Trailing comma must not produce extra entry");
    }
}
```

After creating `src/body_parser.rs`, add `pub mod body_parser;` to `src/lib.rs` (after the existing `pub mod parse;` line). Do NOT add it behind a feature gate — body_parser.rs uses only `crate::model` and `crate::parse::ParseError`, both available under the default feature.

IMPORTANT: The `#[should_panic]` tests are intentional Wave 0 stubs — they document what the implementation must produce and will turn green in Plan 02. The `parse_relationships_empty_body` test body is left empty intentionally (calling `todo!()` would panic the test runner before Plan 02 is ready).
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test body_parser 2>&1 | tail -20</automated>
  </verify>
  <done>
    - `src/body_parser.rs` exists and compiles without errors
    - `pub mod body_parser;` added to `src/lib.rs`
    - `cargo test body_parser` runs (all `#[should_panic]` tests pass because the stubs panic via `todo!()`)
    - Module is visible and test framework confirms it exists
  </done>
</task>

<task type="auto">
  <name>Task 3: Create sqllogictest file and extend parse_proptest.rs with AS-body blocks</name>
  <files>test/sql/phase25_keyword_body.test, tests/parse_proptest.rs</files>
  <action>
**Create `test/sql/phase25_keyword_body.test`:**

This file covers DDL-01 (AS keyword body creates view) and DDL-07 (all 7 verbs work). Use `skipif` to guard tests that need the implementation (Plan 02/03) — but write the structure now so execution test failures are visible.

```sql
# Phase 25: SQL keyword body parser integration tests
# Tests for CREATE SEMANTIC VIEW ... AS TABLES (...) RELATIONSHIPS (...) DIMENSIONS (...) METRICS (...)

# Setup: minimal test table
statement ok
CREATE TABLE IF NOT EXISTS orders (o_orderkey INTEGER, o_custkey INTEGER, revenue DOUBLE, region VARCHAR);

statement ok
CREATE TABLE IF NOT EXISTS customers (c_custkey INTEGER, c_name VARCHAR);

# DDL-01: CREATE SEMANTIC VIEW with keyword body (single table, no relationships)
statement ok
CREATE SEMANTIC VIEW orders_view AS
TABLES (
    o AS orders PRIMARY KEY (o_orderkey)
)
DIMENSIONS (
    o.region AS region
)
METRICS (
    o.total_revenue AS SUM(revenue)
)

# DDL-01: Query the created view to confirm it works
query I rowsort
SELECT total_revenue FROM semantic_view('orders_view', dimensions := [], metrics := ['total_revenue'])
----
NULL

# DDL-07: CREATE OR REPLACE works with keyword body
statement ok
CREATE OR REPLACE SEMANTIC VIEW orders_view AS
TABLES (
    o AS orders PRIMARY KEY (o_orderkey)
)
DIMENSIONS (
    o.region AS region
)
METRICS (
    o.total_revenue AS SUM(revenue)
)

# DDL-07: CREATE IF NOT EXISTS works with keyword body
statement ok
CREATE SEMANTIC VIEW IF NOT EXISTS orders_view AS
TABLES (
    o AS orders PRIMARY KEY (o_orderkey)
)
DIMENSIONS (
    o.region AS region
)
METRICS (
    o.total_revenue AS SUM(revenue)
)

# DDL-07: Multi-table view with RELATIONSHIPS clause
statement ok
CREATE SEMANTIC VIEW orders_customers_view AS
TABLES (
    o AS orders PRIMARY KEY (o_orderkey),
    c AS customers PRIMARY KEY (c_custkey)
)
RELATIONSHIPS (
    order_to_customer AS o(o_custkey) REFERENCES c
)
DIMENSIONS (
    o.region AS region,
    c.c_name AS customer_name
)
METRICS (
    o.total_revenue AS SUM(revenue)
)

# DDL-07: DESCRIBE works
query IIIIII rowsort
DESCRIBE SEMANTIC VIEW orders_view
----

# DDL-07: SHOW works
query II rowsort
SHOW SEMANTIC VIEWS
----

# DDL-07: DROP works
statement ok
DROP SEMANTIC VIEW orders_customers_view

# DDL-07: DROP IF EXISTS works
statement ok
DROP SEMANTIC VIEW IF EXISTS orders_customers_view

# Error: TABLES clause is required
statement error
CREATE SEMANTIC VIEW bad_view AS
DIMENSIONS (
    o.region AS region
)
METRICS (
    o.revenue AS SUM(revenue)
)

# Error: DIMENSIONS or METRICS is required
statement error
CREATE SEMANTIC VIEW bad_view2 AS
TABLES (
    o AS orders PRIMARY KEY (o_orderkey)
)

# Cleanup
statement ok
DROP SEMANTIC VIEW IF EXISTS orders_view
```

**Extend `tests/parse_proptest.rs`:**

Add the following new proptest block after the existing TEST-05 section (after line 655):

```rust
// ---------------------------------------------------------------------------
// AS-body keyword syntax properties (TEST-06)
// ---------------------------------------------------------------------------

/// Build a minimal valid AS-body suffix: AS TABLES (...) DIMENSIONS (...) METRICS (...)
fn build_as_body_suffix(name: &str) -> String {
    format!(
        " {name} AS TABLES (t AS orders PRIMARY KEY (id)) DIMENSIONS (t.region AS region) METRICS (t.revenue AS SUM(amount))"
    )
}

proptest! {
    /// AS-body CREATE forms are detected by detect_ddl_kind.
    #[test]
    fn as_body_detected_for_create_forms(
        form_idx in 0..3usize,
        name in arb_view_name(),
    ) {
        let (prefix, expected_kind, _) = CREATE_FORMS[form_idx];
        let query = format!("{prefix}{}", build_as_body_suffix(&name));
        prop_assert_eq!(
            detect_ddl_kind(&query),
            Some(expected_kind),
            "AS-body DDL should be detected for form: {prefix}"
        );
    }

    /// AS-body validate_and_rewrite returns Ok(Some(_)) for valid keyword body.
    /// This test will FAIL until Plan 03 wires the AS dispatch into validate_create_body.
    /// Expected failure documented here as part of Wave 0 gap contract.
    #[test]
    fn as_body_validate_and_rewrite_succeeds(
        name in arb_view_name(),
    ) {
        let query = format!("CREATE SEMANTIC VIEW{}", build_as_body_suffix(&name));
        // This should succeed after Plan 03 implements the AS dispatch.
        // For now, we assert the kind is detected (weaker check that always passes).
        prop_assert_eq!(detect_ddl_kind(&query), Some(DdlKind::Create));
        // The full validate_and_rewrite check is added in Plan 03 summary test update.
    }

    /// Error position inside AS-body clause points at correct byte offset.
    /// This property verifies the base_offset threading invariant from RESEARCH.md Pitfall 1.
    #[test]
    fn as_body_position_invariant_clause_typo(
        spaces in "[ ]{0,20}",
    ) {
        // When the implementation is complete, this error query should produce
        // an error pointing at "TABLSE". Until Plan 02/03, validate_and_rewrite
        // may return Err with Expected '(' message. We assert the query is detected.
        let query = format!("{spaces}CREATE SEMANTIC VIEW x AS TABLSE (t AS orders PRIMARY KEY (id)) DIMENSIONS (t.r AS r) METRICS (t.m AS SUM(1))");
        prop_assert_eq!(
            detect_ddl_kind(&query),
            Some(DdlKind::Create),
            "AS-body query should be detected even with clause typo in body"
        );
    }
}
```

Add `build_as_body_suffix` as a module-level function (not inside a proptest! block). Add this after the `build_suffix` function at the top of the file.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test parse_proptest 2>&1 | tail -15</automated>
  </verify>
  <done>
    - `test/sql/phase25_keyword_body.test` exists with all 7 DDL verb tests
    - `tests/parse_proptest.rs` has the TEST-06 AS-body proptest block
    - `cargo test parse_proptest` runs (existing tests still pass; new TEST-06 tests pass at the detection level)
    - sqllogictest file is syntactically correct (manually verified by reading it)
  </done>
</task>

</tasks>

<verification>
```bash
cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test 2>&1 | tail -10
cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && just build 2>&1 | tail -5
```
All Rust tests pass. Extension builds successfully with the larger C++ buffer.
</verification>

<success_criteria>
- `just build` succeeds (C++ buffer fix compiles)
- `cargo test` passes — all existing tests green, new body_parser stubs pass via `#[should_panic]`
- `test/sql/phase25_keyword_body.test` exists and is syntactically valid
- `tests/parse_proptest.rs` has TEST-06 AS-body proptest block with detection-level assertions
- No `char sql_buf[4096]` remains in `cpp/src/shim.cpp`
</success_criteria>

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