---
phase: 52-yaml-ddl-integration
plan: 01
type: execute
wave: 1
depends_on: []
files_modified:
  - src/parse.rs
  - test/sql/phase52_yaml_ddl.test
  - test/sql/TEST_LIST
autonomous: true
requirements: [YAML-01, YAML-06]

must_haves:
  truths:
    - "CREATE SEMANTIC VIEW name FROM YAML $$ yaml $$ creates a queryable semantic view"
    - "CREATE OR REPLACE SEMANTIC VIEW name FROM YAML $$ yaml $$ replaces an existing view"
    - "CREATE SEMANTIC VIEW IF NOT EXISTS name FROM YAML $$ yaml $$ is a no-op when the view already exists"
    - "Invalid YAML in a dollar-quoted block returns a clear error mentioning the view name"
    - "The error message for missing body mentions both AS and FROM YAML syntax"
    - "Trailing content after closing $$ is rejected as a parse error"
    - "Tagged dollar-quoting ($yaml$...$yaml$) works identically to untagged $$...$$"
  artifacts:
    - path: "src/parse.rs"
      provides: "extract_dollar_quoted, rewrite_ddl_yaml_body, FROM YAML detection in validate_create_body"
      contains: "fn extract_dollar_quoted"
    - path: "src/parse.rs"
      provides: "YAML rewrite function"
      contains: "fn rewrite_ddl_yaml_body"
    - path: "test/sql/phase52_yaml_ddl.test"
      provides: "sqllogictest integration tests for YAML DDL"
      contains: "FROM YAML"
    - path: "test/sql/TEST_LIST"
      provides: "updated test registry"
      contains: "phase52_yaml_ddl.test"
  key_links:
    - from: "src/parse.rs::validate_create_body"
      to: "src/parse.rs::rewrite_ddl_yaml_body"
      via: "FROM YAML detection branch"
      pattern: "is_yaml_body.*rewrite_ddl_yaml_body"
    - from: "src/parse.rs::rewrite_ddl_yaml_body"
      to: "src/model.rs::from_yaml_with_size_cap"
      via: "YAML deserialization call"
      pattern: "from_yaml_with_size_cap"
    - from: "src/parse.rs::rewrite_ddl_yaml_body"
      to: "src/parse.rs::infer_cardinality"
      via: "cardinality inference (matching SQL DDL path)"
      pattern: "infer_cardinality"
---

<objective>
Wire the YAML parsing capability (from Phase 51) into the DDL pipeline so that
`CREATE SEMANTIC VIEW name FROM YAML $$ ... $$` creates a queryable semantic view.

Purpose: Enable users to define semantic views using YAML syntax as an alternative to
SQL DDL keyword body. The YAML path converges to the same JSON-parameterized function
call as the SQL path, reusing all existing validation, persistence, and query infrastructure.

Output: Modified `src/parse.rs` with FROM YAML detection, dollar-quote extraction, and
YAML-to-JSON rewrite function; comprehensive unit tests in parse.rs `mod tests`;
sqllogictest integration tests in `test/sql/phase52_yaml_ddl.test`.
</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/52-yaml-ddl-integration/52-RESEARCH.md
@.planning/phases/51-yaml-parser-core/51-01-SUMMARY.md

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

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

From src/model.rs (Phase 51):
```rust
impl SemanticViewDefinition {
    pub const YAML_SIZE_CAP: usize = 1_048_576;
    pub fn from_yaml(name: &str, yaml: &str) -> Result<Self, String>;
    pub fn from_yaml_with_size_cap(name: &str, yaml: &str) -> Result<Self, String>;
}
```

From src/parse.rs (existing):
```rust
pub enum DdlKind {
    Create, CreateOrReplace, CreateIfNotExists,
    Drop, DropIfExists,
    Alter, AlterIfExists,
    Describe, Show, ShowColumns, ShowTerse,
    ShowDimensions, ShowMetrics, ShowFacts,
}

fn validate_create_body(
    _query: &str, trimmed_no_semi: &str, trim_offset: usize,
    plen: usize, kind: DdlKind,
) -> Result<Option<String>, ParseError>;

fn rewrite_ddl_keyword_body(
    kind: DdlKind, name: &str, body_text: &str,
    body_offset: usize, view_comment: Option<String>,
) -> Result<Option<String>, ParseError>;

pub(crate) fn infer_cardinality(
    tables: &[TableRef], relationships: &mut [Join],
) -> Result<(), ParseError>;

fn extract_view_comment(text: &str) -> Result<(Option<String>, &str), ParseError>;
```
</interfaces>
</context>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Add dollar-quote extraction, YAML rewrite, and FROM YAML detection to parse.rs</name>
  <files>src/parse.rs</files>
  <read_first>
    - src/parse.rs (lines 996-1125: validate_create_body and rewrite_ddl_keyword_body — the insertion point and the pattern to mirror)
    - src/model.rs (lines 427-457: from_yaml_with_size_cap signature and behavior)
    - src/errors.rs (ParseError struct)
  </read_first>
  <behavior>
    - Test: extract_dollar_quoted("$$hello$$") returns Ok(("hello", 11))
    - Test: extract_dollar_quoted("$yaml$content$yaml$") returns Ok(("content", 19))
    - Test: extract_dollar_quoted("$$") returns Err (unterminated — no content or closing delimiter)
    - Test: extract_dollar_quoted("not a dollar") returns Err (no leading $)
    - Test: extract_dollar_quoted("$$has inner $$ text$$") returns Ok(("has inner ", 14)) — first closing $$ wins
    - Test: rewrite_ddl_yaml_body with valid YAML returns Ok(Some("SELECT * FROM create_semantic_view_from_json(...)"))
    - Test: rewrite_ddl_yaml_body with DdlKind::CreateOrReplace uses create_or_replace_semantic_view_from_json
    - Test: rewrite_ddl_yaml_body with DdlKind::CreateIfNotExists uses create_semantic_view_if_not_exists_from_json
    - Test: rewrite_ddl_yaml_body with trailing content after closing $$ returns Err
    - Test: rewrite_ddl_yaml_body with invalid YAML returns Err with view name in message
    - Test: validate_create_body routes FROM YAML path correctly (via rewrite_ddl)
    - Test: FROM YAML detection is case-insensitive (from yaml, From Yaml, FROM YAML)
    - Test: error message for missing body mentions both AS and FROM YAML
    - Test: COMMENT = '...' FROM YAML $$...$$ sets comment (DDL comment overrides YAML comment)
    - Test: YAML with empty base_table gets it populated from first table entry
  </behavior>
  <action>
Add three new functions and modify one existing function in src/parse.rs:

**1. Add `extract_dollar_quoted` function** (insert after line 1125, before the cardinality inference section):

```rust
/// Extract content from a dollar-quoted string (`$$...$$` or `$tag$...$tag$`).
///
/// Returns `(content, bytes_consumed)` where bytes_consumed includes both
/// opening and closing delimiters. The content does NOT include the delimiters.
fn extract_dollar_quoted(input: &str) -> Result<(String, usize), ParseError> {
    if !input.starts_with('$') {
        return Err(ParseError {
            message: "Expected '$' to begin dollar-quoted string".to_string(),
            position: None,
        });
    }
    let tag_end = input[1..]
        .find('$')
        .ok_or_else(|| ParseError {
            message: "Unterminated dollar-quote opening delimiter".to_string(),
            position: None,
        })?
        + 2;
    let delimiter = &input[..tag_end];
    let content_start = tag_end;
    let close_pos = input[content_start..].find(delimiter).ok_or_else(|| ParseError {
        message: format!(
            "Unterminated dollar-quoted string (expected closing '{delimiter}')"
        ),
        position: None,
    })?;
    let content = &input[content_start..content_start + close_pos];
    let total = content_start + close_pos + delimiter.len();
    Ok((content.to_string(), total))
}
```

**2. Add `rewrite_ddl_yaml_body` function** (insert immediately after `extract_dollar_quoted`):

```rust
/// Rewrite a FROM YAML dollar-quoted DDL statement to a JSON-parameterized function call.
///
/// Called when `validate_create_body` detects the `FROM YAML` keyword path.
/// Extracts dollar-quoted YAML, deserializes via `from_yaml_with_size_cap()`,
/// serializes to JSON, and embeds in a `SELECT * FROM create_semantic_view_from_json('name', 'json')` call.
fn rewrite_ddl_yaml_body(
    kind: DdlKind,
    name: &str,
    yaml_text: &str,
    view_comment: Option<String>,
) -> Result<Option<String>, ParseError> {
    let (yaml_content, consumed) = extract_dollar_quoted(yaml_text)?;

    let trailing = yaml_text[consumed..].trim();
    if !trailing.is_empty() {
        return Err(ParseError {
            message: format!("Unexpected content after closing dollar-quote: '{trailing}'"),
            position: None,
        });
    }

    let mut def = crate::model::SemanticViewDefinition::from_yaml_with_size_cap(name, &yaml_content)
        .map_err(|e| ParseError {
            message: e,
            position: None,
        })?;

    if let Some(c) = view_comment {
        def.comment = Some(c);
    }

    if def.base_table.is_empty() {
        if let Some(first) = def.tables.first() {
            def.base_table = first.table.clone();
        }
    }

    infer_cardinality(&def.tables, &mut def.joins)?;

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

    let safe_name = name.replace('\'', "''");
    let safe_json = json.replace('\'', "''");
    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_yaml_body only called for CREATE forms"),
    };
    Ok(Some(format!(
        "SELECT * FROM {fn_name}('{safe_name}', '{safe_json}')"
    )))
}
```

**3. Modify `validate_create_body`** — Insert the FROM YAML detection block between the end of the AS body path (after line 1052 `// --- End AS keyword body path ---`) and before the error at line 1055. The new block:

```rust
// --- FROM YAML body path (Phase 52) ---
let is_yaml_body = after_name_trimmed
    .get(..9)
    .is_some_and(|s| s.eq_ignore_ascii_case("FROM YAML"))
    && (after_name_trimmed.len() == 9
        || after_name_trimmed.as_bytes()[9].is_ascii_whitespace());
if is_yaml_body {
    let yaml_text = after_name_trimmed[9..].trim_start();
    return rewrite_ddl_yaml_body(kind, name, yaml_text, view_comment);
}
// --- End FROM YAML body path ---
```

**4. Update the error message** at the bottom of `validate_create_body` (currently line 1058) to mention both syntaxes:

Change from:
```
"Expected 'AS' keyword after view name. Use: CREATE SEMANTIC VIEW name AS TABLES (...) DIMENSIONS (...) METRICS (...)"
```

To:
```
"Expected 'AS' or 'FROM YAML' after view name. Use: CREATE SEMANTIC VIEW name AS TABLES (...) DIMENSIONS (...) METRICS (...) or: CREATE SEMANTIC VIEW name FROM YAML $$ ... $$"
```

**5. Add unit tests** in the existing `mod tests` block at the end of parse.rs. Add a new section:

```rust
// ===================================================================
// Phase 52: Dollar-quote extraction tests
// ===================================================================

#[test]
fn test_extract_dollar_quoted_untagged() {
    let (content, consumed) = extract_dollar_quoted("$$hello world$$").unwrap();
    assert_eq!(content, "hello world");
    assert_eq!(consumed, 15);
}

#[test]
fn test_extract_dollar_quoted_tagged() {
    let (content, consumed) = extract_dollar_quoted("$yaml$my content$yaml$").unwrap();
    assert_eq!(content, "my content");
    assert_eq!(consumed, 22);
}

#[test]
fn test_extract_dollar_quoted_empty_content() {
    let (content, consumed) = extract_dollar_quoted("$$$$").unwrap();
    assert_eq!(content, "");
    assert_eq!(consumed, 4);
}

#[test]
fn test_extract_dollar_quoted_no_leading_dollar() {
    let err = extract_dollar_quoted("not a dollar").unwrap_err();
    assert!(err.message.contains("Expected '$'"));
}

#[test]
fn test_extract_dollar_quoted_unterminated_opening() {
    let err = extract_dollar_quoted("$no_close").unwrap_err();
    assert!(err.message.contains("Unterminated dollar-quote opening"));
}

#[test]
fn test_extract_dollar_quoted_unterminated_body() {
    let err = extract_dollar_quoted("$$no closing").unwrap_err();
    assert!(err.message.contains("Unterminated dollar-quoted string"));
}

#[test]
fn test_extract_dollar_quoted_inner_dollar() {
    // First closing $$ wins — content is "has inner "
    let (content, consumed) = extract_dollar_quoted("$$has inner $$ text$$").unwrap();
    assert_eq!(content, "has inner ");
    assert_eq!(consumed, 14);
}

#[test]
fn test_extract_dollar_quoted_multiline() {
    let input = "$$\ntables:\n  - alias: o\n    table: orders\n$$";
    let (content, _) = extract_dollar_quoted(input).unwrap();
    assert!(content.contains("tables:"));
    assert!(content.contains("alias: o"));
}

// ===================================================================
// Phase 52: YAML DDL rewrite tests
// ===================================================================

#[test]
fn test_yaml_rewrite_basic_create() {
    let yaml_text = r#"$$
base_table: orders
tables:
  - alias: o
    table: orders
    pk_columns:
      - id
dimensions:
  - name: region
    expr: o.region
    source_table: o
metrics:
  - name: total_amount
    expr: SUM(o.amount)
    source_table: o
$$"#;
    let result = rewrite_ddl_yaml_body(DdlKind::Create, "test_view", yaml_text, None).unwrap();
    let sql = result.unwrap();
    assert!(sql.starts_with("SELECT * FROM create_semantic_view_from_json('test_view',"));
}

#[test]
fn test_yaml_rewrite_create_or_replace() {
    let yaml_text = "$$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$";
    let result = rewrite_ddl_yaml_body(DdlKind::CreateOrReplace, "v", yaml_text, None).unwrap();
    let sql = result.unwrap();
    assert!(sql.contains("create_or_replace_semantic_view_from_json"));
}

#[test]
fn test_yaml_rewrite_create_if_not_exists() {
    let yaml_text = "$$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$";
    let result = rewrite_ddl_yaml_body(DdlKind::CreateIfNotExists, "v", yaml_text, None).unwrap();
    let sql = result.unwrap();
    assert!(sql.contains("create_semantic_view_if_not_exists_from_json"));
}

#[test]
fn test_yaml_rewrite_trailing_content_rejected() {
    let yaml_text = "$$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$ extra stuff";
    let err = rewrite_ddl_yaml_body(DdlKind::Create, "v", yaml_text, None).unwrap_err();
    assert!(err.message.contains("Unexpected content after closing dollar-quote"));
}

#[test]
fn test_yaml_rewrite_invalid_yaml() {
    let yaml_text = "$$\n: : : not valid yaml [[[$$";
    let err = rewrite_ddl_yaml_body(DdlKind::Create, "bad_view", yaml_text, None).unwrap_err();
    assert!(err.message.contains("bad_view"));
}

#[test]
fn test_yaml_rewrite_comment_override() {
    let yaml_text = "$$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\ncomment: yaml comment\n$$";
    let result = rewrite_ddl_yaml_body(DdlKind::Create, "v", yaml_text, Some("ddl comment".to_string())).unwrap();
    let sql = result.unwrap();
    // DDL comment overrides YAML comment
    assert!(sql.contains("ddl comment"));
}

#[test]
fn test_yaml_rewrite_base_table_populated() {
    let yaml_text = r#"$$
base_table: ""
tables:
  - alias: o
    table: orders
    pk_columns: []
dimensions: []
metrics: []
$$"#;
    let result = rewrite_ddl_yaml_body(DdlKind::Create, "v", yaml_text, None).unwrap();
    let sql = result.unwrap();
    // base_table should be populated from first table entry
    assert!(sql.contains("orders"));
}

#[test]
fn test_yaml_rewrite_tagged_dollar_quote() {
    let yaml_text = "$yaml$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$yaml$";
    let result = rewrite_ddl_yaml_body(DdlKind::Create, "v", yaml_text, None).unwrap();
    assert!(result.is_some());
}

// ===================================================================
// Phase 52: FROM YAML detection in validate_create_body
// ===================================================================

#[test]
fn test_from_yaml_detection_via_rewrite_ddl() {
    let query = r#"CREATE SEMANTIC VIEW yaml_test FROM YAML $$
base_table: t
tables: []
dimensions: []
metrics: []
$$"#;
    let result = rewrite_ddl(query).unwrap();
    assert!(result.is_some());
    let sql = result.unwrap();
    assert!(sql.contains("create_semantic_view_from_json('yaml_test'"));
}

#[test]
fn test_from_yaml_case_insensitive() {
    let query = "CREATE SEMANTIC VIEW v from yaml $$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$";
    let result = rewrite_ddl(query).unwrap();
    assert!(result.is_some());
}

#[test]
fn test_from_yaml_mixed_case() {
    let query = "CREATE SEMANTIC VIEW v From Yaml $$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$";
    let result = rewrite_ddl(query).unwrap();
    assert!(result.is_some());
}

#[test]
fn test_error_message_mentions_from_yaml() {
    let query = "CREATE SEMANTIC VIEW v SOMETHING_ELSE";
    let err = rewrite_ddl(query).unwrap_err();
    assert!(err.message.contains("FROM YAML"), "Error should mention FROM YAML: {}", err.message);
}

#[test]
fn test_create_or_replace_from_yaml() {
    let query = "CREATE OR REPLACE SEMANTIC VIEW v FROM YAML $$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$";
    let result = rewrite_ddl(query).unwrap();
    let sql = result.unwrap();
    assert!(sql.contains("create_or_replace_semantic_view_from_json"));
}

#[test]
fn test_create_if_not_exists_from_yaml() {
    let query = "CREATE SEMANTIC VIEW IF NOT EXISTS v FROM YAML $$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$";
    let result = rewrite_ddl(query).unwrap();
    let sql = result.unwrap();
    assert!(sql.contains("create_semantic_view_if_not_exists_from_json"));
}

#[test]
fn test_comment_with_from_yaml() {
    let query = "CREATE SEMANTIC VIEW v COMMENT = 'my comment' FROM YAML $$\nbase_table: t\ntables: []\ndimensions: []\nmetrics: []\n$$";
    let result = rewrite_ddl(query).unwrap();
    let sql = result.unwrap();
    assert!(sql.contains("my comment"));
}
```
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test parse::tests::yaml -- --nocapture 2>&amp;1 | tail -5</automated>
  </verify>
  <acceptance_criteria>
    - src/parse.rs contains `fn extract_dollar_quoted(input: &str) -> Result<(String, usize), ParseError>`
    - src/parse.rs contains `fn rewrite_ddl_yaml_body(`
    - src/parse.rs contains `eq_ignore_ascii_case("FROM YAML")`
    - src/parse.rs contains `"Expected 'AS' or 'FROM YAML' after view name`
    - src/parse.rs contains `from_yaml_with_size_cap`
    - src/parse.rs contains `infer_cardinality(&def.tables, &mut def.joins)`
    - src/parse.rs contains `fn test_extract_dollar_quoted_untagged`
    - src/parse.rs contains `fn test_yaml_rewrite_basic_create`
    - src/parse.rs contains `fn test_from_yaml_detection_via_rewrite_ddl`
    - src/parse.rs contains `fn test_from_yaml_case_insensitive`
    - src/parse.rs contains `fn test_error_message_mentions_from_yaml`
    - `cargo test parse::tests` exits 0
    - `cargo test` exits 0 (full unit test suite including existing tests still passes)
  </acceptance_criteria>
  <done>
    All three new functions (extract_dollar_quoted, rewrite_ddl_yaml_body, FROM YAML detection)
    are implemented in src/parse.rs. Unit tests cover: dollar-quote extraction (untagged, tagged,
    empty, unterminated, multiline), YAML rewrite (all 3 CREATE variants, trailing content rejection,
    invalid YAML, comment override, base_table population, tagged delimiters), and FROM YAML detection
    (case-insensitivity, error message update, all DDL modifiers, COMMENT integration).
    All existing tests continue to pass.
  </done>
</task>

<task type="auto">
  <name>Task 2: Add sqllogictest integration tests and update TEST_LIST</name>
  <files>test/sql/phase52_yaml_ddl.test, test/sql/TEST_LIST</files>
  <read_first>
    - test/sql/phase25_keyword_body.test (lines 1-30: existing sqllogictest pattern for DDL integration tests — require, setup, statement ok/error, query assertions)
    - test/sql/TEST_LIST (current test file list — new file must be appended)
    - src/parse.rs (the FROM YAML detection and rewrite code from Task 1, to understand expected behavior)
  </read_first>
  <action>
**1. Create `test/sql/phase52_yaml_ddl.test`** with the following complete content:

```sql
# Phase 52: YAML DDL integration tests
# Tests for CREATE SEMANTIC VIEW ... FROM YAML $$ ... $$

require semantic_views

# Setup: test tables with data
statement ok
CREATE TABLE p52_orders (id INTEGER PRIMARY KEY, amount DOUBLE, region VARCHAR);

statement ok
INSERT INTO p52_orders VALUES (1, 100.0, 'East'), (2, 200.0, 'West'), (3, 150.0, 'East');

statement ok
CREATE TABLE p52_customers (cust_id INTEGER PRIMARY KEY, name VARCHAR);

statement ok
INSERT INTO p52_customers VALUES (10, 'Alice'), (20, 'Bob');

# ---------------------------------------------------------------
# YAML-01: CREATE SEMANTIC VIEW from inline YAML
# ---------------------------------------------------------------

statement ok
CREATE SEMANTIC VIEW p52_yaml_basic FROM YAML $$
base_table: p52_orders
tables:
  - alias: o
    table: p52_orders
    pk_columns:
      - id
dimensions:
  - name: region
    expr: o.region
    source_table: o
metrics:
  - name: total_amount
    expr: SUM(o.amount)
    source_table: o
$$

# Verify the view is queryable
query TR rowsort
SELECT region, total_amount FROM semantic_view('p52_yaml_basic', dimensions := ['region'], metrics := ['total_amount'])
----
East	250.0
West	200.0

# Verify metrics-only query works
query R
SELECT total_amount FROM semantic_view('p52_yaml_basic', metrics := ['total_amount'])
----
450.0

# ---------------------------------------------------------------
# YAML-01: Tagged dollar-quoting ($yaml$...$yaml$)
# ---------------------------------------------------------------

statement ok
CREATE SEMANTIC VIEW p52_yaml_tagged FROM YAML $yaml$
base_table: p52_orders
tables:
  - alias: o
    table: p52_orders
    pk_columns:
      - id
dimensions:
  - name: region
    expr: o.region
    source_table: o
metrics:
  - name: total_amount
    expr: SUM(o.amount)
    source_table: o
$yaml$

query R
SELECT total_amount FROM semantic_view('p52_yaml_tagged', metrics := ['total_amount'])
----
450.0

# ---------------------------------------------------------------
# YAML-06: CREATE OR REPLACE with FROM YAML
# ---------------------------------------------------------------

statement ok
CREATE OR REPLACE SEMANTIC VIEW p52_yaml_basic FROM YAML $$
base_table: p52_orders
tables:
  - alias: o
    table: p52_orders
    pk_columns:
      - id
dimensions:
  - name: region
    expr: o.region
    source_table: o
metrics:
  - name: order_count
    expr: COUNT(o.id)
    source_table: o
$$

# Verify the view was replaced (new metric name)
query TR rowsort
SELECT region, order_count FROM semantic_view('p52_yaml_basic', dimensions := ['region'], metrics := ['order_count'])
----
East	2
West	1

# ---------------------------------------------------------------
# YAML-06: CREATE IF NOT EXISTS (no-op when view exists)
# ---------------------------------------------------------------

statement ok
CREATE SEMANTIC VIEW IF NOT EXISTS p52_yaml_basic FROM YAML $$
base_table: p52_orders
tables:
  - alias: o
    table: p52_orders
    pk_columns:
      - id
dimensions:
  - name: region
    expr: o.region
    source_table: o
metrics:
  - name: should_not_appear
    expr: SUM(o.amount)
    source_table: o
$$

# Verify the original view (with order_count) is still in place
query TR rowsort
SELECT region, order_count FROM semantic_view('p52_yaml_basic', dimensions := ['region'], metrics := ['order_count'])
----
East	2
West	1

# ---------------------------------------------------------------
# YAML-01: COMMENT = '...' with FROM YAML
# ---------------------------------------------------------------

statement ok
CREATE SEMANTIC VIEW p52_yaml_commented COMMENT = 'Sales by region from YAML' FROM YAML $$
base_table: p52_orders
tables:
  - alias: o
    table: p52_orders
    pk_columns:
      - id
dimensions:
  - name: region
    expr: o.region
    source_table: o
metrics:
  - name: total_amount
    expr: SUM(o.amount)
    source_table: o
$$

query R
SELECT total_amount FROM semantic_view('p52_yaml_commented', metrics := ['total_amount'])
----
450.0

# ---------------------------------------------------------------
# YAML-01: Case-insensitive FROM YAML detection
# ---------------------------------------------------------------

statement ok
CREATE SEMANTIC VIEW p52_yaml_lowercase from yaml $$
base_table: p52_orders
tables:
  - alias: o
    table: p52_orders
    pk_columns:
      - id
dimensions:
  - name: region
    expr: o.region
    source_table: o
metrics:
  - name: total_amount
    expr: SUM(o.amount)
    source_table: o
$$

query R
SELECT total_amount FROM semantic_view('p52_yaml_lowercase', metrics := ['total_amount'])
----
450.0

# ---------------------------------------------------------------
# Error cases
# ---------------------------------------------------------------

# Invalid YAML content
statement error
CREATE SEMANTIC VIEW p52_bad_yaml FROM YAML $$
: : : not valid yaml [[[
$$
----
invalid YAML definition

# Unterminated dollar-quote
statement error
CREATE SEMANTIC VIEW p52_unterminated FROM YAML $$
base_table: orders
----
Unterminated dollar-quoted string

# Missing dollar-quote entirely
statement error
CREATE SEMANTIC VIEW p52_no_dollar FROM YAML base_table: orders
----
Expected '$'

# ---------------------------------------------------------------
# Cleanup
# ---------------------------------------------------------------

statement ok
DROP SEMANTIC VIEW p52_yaml_basic

statement ok
DROP SEMANTIC VIEW p52_yaml_tagged

statement ok
DROP SEMANTIC VIEW p52_yaml_commented

statement ok
DROP SEMANTIC VIEW p52_yaml_lowercase

statement ok
DROP TABLE p52_orders

statement ok
DROP TABLE p52_customers
```

**2. Update `test/sql/TEST_LIST`** — Append the following line at the end of the file:

```
test/sql/phase52_yaml_ddl.test
```

Append after the last existing entry (`test/sql/peg_compat.test`).
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && just build && just test-sql 2>&amp;1 | tail -10</automated>
  </verify>
  <acceptance_criteria>
    - test/sql/phase52_yaml_ddl.test exists and contains `require semantic_views`
    - test/sql/phase52_yaml_ddl.test contains `CREATE SEMANTIC VIEW p52_yaml_basic FROM YAML $$`
    - test/sql/phase52_yaml_ddl.test contains `CREATE OR REPLACE SEMANTIC VIEW p52_yaml_basic FROM YAML $$`
    - test/sql/phase52_yaml_ddl.test contains `CREATE SEMANTIC VIEW IF NOT EXISTS p52_yaml_basic FROM YAML $$`
    - test/sql/phase52_yaml_ddl.test contains `$yaml$` (tagged dollar-quoting test)
    - test/sql/phase52_yaml_ddl.test contains `from yaml` (case-insensitive test)
    - test/sql/phase52_yaml_ddl.test contains `COMMENT = 'Sales by region from YAML'`
    - test/sql/phase52_yaml_ddl.test contains `statement error` (at least 3 error test cases)
    - test/sql/TEST_LIST contains `test/sql/phase52_yaml_ddl.test`
    - `just build` exits 0
    - `just test-sql` exits 0 (all sqllogictest files pass including the new one)
    - `just test-all` exits 0 (full quality gate)
  </acceptance_criteria>
  <done>
    Sqllogictest integration tests exercise the full YAML DDL pipeline end-to-end through the
    loaded extension: basic CREATE FROM YAML, tagged dollar-quoting, CREATE OR REPLACE, CREATE IF
    NOT EXISTS, COMMENT integration, case-insensitive detection, and three error cases (invalid YAML,
    unterminated dollar-quote, missing dollar-quote). All tests pass under `just test-all`.
    TEST_LIST updated with the new test file.
  </done>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| DDL input -> parser | User-supplied YAML content in dollar-quoted block crosses into deserialization |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-52-01 | Denial of Service | rewrite_ddl_yaml_body | mitigate | 1MB size cap enforced via `from_yaml_with_size_cap()` before YAML parsing begins (inherited from Phase 51) |
| T-52-02 | Tampering | extract_dollar_quoted | mitigate | Trailing content after closing delimiter rejected with ParseError; prevents injection of extra SQL after YAML block |
| T-52-03 | Tampering | rewrite_ddl_yaml_body | mitigate | YAML deserialized into typed Rust structs via serde, then re-serialized to JSON. Single quotes SQL-escaped via `replace('\'', "''")`. No raw YAML text passes through SQL. |
| T-52-04 | Denial of Service | extract_dollar_quoted | accept | Unterminated dollar-quote scans the full input string. Input is bounded by the 1MB YAML size cap + DDL prefix overhead. Acceptable cost for a CREATE DDL operation. |
</threat_model>

<verification>
1. `cargo test parse::tests` — all unit tests pass (existing + new Phase 52 tests)
2. `cargo test` — full Rust test suite passes (including Phase 51 YAML tests, proptests)
3. `just build && just test-sql` — all sqllogictest files pass including phase52_yaml_ddl.test
4. `just test-all` — complete quality gate (Rust + sqllogictest + DuckLake CI)
</verification>

<success_criteria>
- `CREATE SEMANTIC VIEW name FROM YAML $$ yaml $$ ` creates a semantic view queryable via `semantic_view()`
- `CREATE OR REPLACE SEMANTIC VIEW name FROM YAML $$ yaml $$` replaces an existing view
- `CREATE SEMANTIC VIEW IF NOT EXISTS name FROM YAML $$ yaml $$` is a no-op for existing views
- Tagged dollar-quoting (`$yaml$...$yaml$`) works identically to untagged `$$...$$`
- Invalid YAML, unterminated dollar-quotes, and trailing content produce clear error messages
- COMMENT = '...' FROM YAML is supported with DDL comment overriding YAML comment field
- `just test-all` passes with zero failures
</success_criteria>

<output>
After completion, create `.planning/phases/52-yaml-ddl-integration/52-01-SUMMARY.md`
</output>
