---
phase: 10-pragma-query-t-catalog-persistence
plan: 02
type: execute
wave: 2
depends_on: [10-01]
files_modified:
  - src/ddl/define.rs
  - src/ddl/drop.rs
  - src/lib.rs
autonomous: true
requirements: [PERSIST-01, PERSIST-02]

must_haves:
  truths:
    - "DefineState no longer has a db_path field — it has a persist_conn: Option<duckdb_connection> field instead"
    - "DropState no longer has a db_path field — it has persist_conn: Option<duckdb_connection> instead"
    - "define_semantic_view invoke calls semantic_views_pragma_define BEFORE catalog_insert (Option B: write-first then HashMap)"
    - "define_semantic_view invoke propagates the FFI error with view name in the message if the pragma write fails"
    - "drop_semantic_view invoke calls semantic_views_pragma_drop BEFORE catalog_delete (write-first ordering)"
    - "lib.rs creates a separate persist_conn via duckdb_connect and passes it to DefineState and DropState"
    - "In-memory databases (persist_conn = None) skip the pragma write path — HashMap only"
    - "cargo test passes with no regressions"
  artifacts:
    - path: "src/ddl/define.rs"
      provides: "Updated DefineState + invoke with pragma FFI write-first pattern"
      contains: "persist_conn"
    - path: "src/ddl/drop.rs"
      provides: "Updated DropState + invoke with pragma FFI write-first pattern"
      contains: "persist_conn"
    - path: "src/lib.rs"
      provides: "persist_conn connection creation at init_extension + passed to DefineState/DropState"
      contains: "persist_conn"
  key_links:
    - from: "src/lib.rs init_extension"
      to: "src/ddl/define.rs DefineState / src/ddl/drop.rs DropState"
      via: "persist_conn: Some(duckdb_connection) passed in state constructor"
      pattern: "persist_conn"
    - from: "src/ddl/define.rs invoke"
      to: "src/shim/mod.rs ffi::semantic_views_pragma_define"
      via: "unsafe call with CString name + json, returns i32 error code"
      pattern: "semantic_views_pragma_define"
    - from: "src/ddl/drop.rs invoke"
      to: "src/shim/mod.rs ffi::semantic_views_pragma_drop"
      via: "unsafe call with CString name, returns i32 error code"
      pattern: "semantic_views_pragma_drop"
---

<objective>
Replace sidecar writes in define/drop invoke with pragma FFI calls, and wire up the persist_conn in lib.rs.

Purpose: After Plan 01 adds the C++ pragma functions, this plan updates the Rust invocation path. `write_sidecar` calls are removed from `define.rs` and `drop.rs`. The new pattern writes to the persistent table via `semantic_views_pragma_define`/`_drop` on a separate stored connection (no deadlock). `lib.rs` creates this connection at init time. The write-first ordering (Option B from CONTEXT.md) ensures HashMap never diverges from a committed table state.

Output: Updated define.rs, drop.rs, and lib.rs. `write_sidecar` is no longer called from invoke — but the sidecar functions themselves are not yet deleted (that happens in Plan 03 after catalog.rs migration).
</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/STATE.md
@.planning/phases/10-pragma-query-t-catalog-persistence/10-RESEARCH.md
@.planning/phases/10-pragma-query-t-catalog-persistence/10-01-SUMMARY.md
@src/ddl/define.rs
@src/ddl/drop.rs
@src/lib.rs
@src/shim/mod.rs

<interfaces>
<!-- Key interfaces needed by this plan -->

From src/shim/mod.rs (after Plan 01 — executor MUST read to confirm exact signatures):
```rust
#[cfg(feature = "extension")]
pub mod ffi {
    use libduckdb_sys::duckdb_connection;
    extern "C" {
        pub fn semantic_views_pragma_define(
            conn: duckdb_connection,
            name: *const std::ffi::c_char,
            json: *const std::ffi::c_char,
        ) -> i32;

        pub fn semantic_views_pragma_drop(
            conn: duckdb_connection,
            name: *const std::ffi::c_char,
        ) -> i32;
    }
}
```

From src/ddl/define.rs (current state — executor MUST read before modifying):
```rust
#[derive(Clone)]
pub struct DefineState {
    pub catalog: CatalogState,
    pub db_path: Arc<str>,  // TO REMOVE
}

// invoke currently: catalog_insert first, then write_sidecar
// NEW order: pragma_define first (if file-backed), then catalog_insert
```

From src/ddl/drop.rs (current state — executor MUST read before modifying):
```rust
#[derive(Clone)]
pub struct DropState {
    pub catalog: CatalogState,
    pub db_path: Arc<str>,  // TO REMOVE
}

// invoke currently: catalog_delete first, then write_sidecar
// NEW order: pragma_drop first (if file-backed), then catalog_delete
```

From src/lib.rs (init_extension — executor MUST read before modifying):
```rust
// Current: creates query_conn for semantic_query, passes db_path to DefineState/DropState
let mut query_conn: ffi::duckdb_connection = ptr::null_mut();
let rc = unsafe { ffi::duckdb_connect(db_handle, &mut query_conn) };

// Phase 10 change: ALSO create persist_conn for DDL writes
// persist_conn is only created for file-backed databases (db_path != ":memory:")
// For :memory:, pass None to DefineState/DropState
```

Raw pointer `Send + Sync` pattern (from src/query/table_function.rs — established pattern):
```rust
// The duckdb_connection raw pointer is Send + Sync per QueryState pattern.
// DefineState and DropState both derive Clone and are shared across threads.
// Wrap the raw pointer in a newtype or use the same unsafe impl pattern.
// Since DefineState is Clone (required by VScalar::State bound), the connection
// is shared across clones — this is safe because DuckDB connections handle
// concurrent access internally.
```

CString pattern for FFI calls (use this for name/json string conversion):
```rust
use std::ffi::CString;

let c_name = CString::new(name.as_str())
    .map_err(|_| format!("view name '{}' contains null byte", name))?;
let c_json = CString::new(json.as_str())
    .map_err(|_| format!("definition JSON contains null byte"))?;

let rc = unsafe {
    crate::shim::ffi::semantic_views_pragma_define(conn, c_name.as_ptr(), c_json.as_ptr())
};
if rc != 0 {
    return Err(format!(
        "failed to persist semantic view '{}': table write failed (rc={})", name, rc
    ).into());
}
```
</interfaces>
</context>

<tasks>

<task type="auto">
  <name>Task 1: Update DefineState and DropState — replace db_path with persist_conn</name>
  <files>src/ddl/define.rs, src/ddl/drop.rs</files>
  <action>
    ### src/ddl/define.rs

    1. Remove the `write_sidecar` import:
       - Old: `use crate::catalog::{catalog_insert, write_sidecar, CatalogState};`
       - New: `use crate::catalog::{catalog_insert, CatalogState};`

    2. Add `use std::ffi::CString;` to the imports.

    3. Replace `DefineState`:
       ```rust
       /// Shared state for `define_semantic_view`.
       ///
       /// `persist_conn` is `Some` for file-backed databases — it is a separate
       /// `duckdb_connection` created at init time and used to execute INSERT into
       /// `semantic_layer._definitions` from within invoke (avoids deadlock with
       /// the main connection's execution lock). For in-memory databases, `persist_conn`
       /// is `None` and the `HashMap` is the sole source of truth for the session.
       #[derive(Clone)]
       pub struct DefineState {
           pub catalog: CatalogState,
           pub persist_conn: Option<libduckdb_sys::duckdb_connection>,
       }

       // SAFETY: duckdb_connection is an opaque pointer managed by DuckDB.
       // DuckDB handles concurrent access internally.
       unsafe impl Send for DefineState {}
       unsafe impl Sync for DefineState {}
       ```

    4. Update `invoke` in `DefineSemanticView`:

       Change the loop body to write-first (Option B from CONTEXT.md):

       ```rust
       for i in 0..input.len() {
           let name = duckdb::types::DuckString::new(&mut { names[i] })
               .as_str()
               .to_string();
           let json = duckdb::types::DuckString::new(&mut { jsons[i] })
               .as_str()
               .to_string();

           // 1. Persist to DuckDB table FIRST (file-backed databases only).
           //    Uses a separate connection — no deadlock with invoke's execution lock.
           //    Write-first ordering: if this fails, HashMap is unchanged (PERSIST-02).
           #[cfg(feature = "extension")]
           if let Some(conn) = state.persist_conn {
               let c_name = CString::new(name.as_str())
                   .map_err(|_| format!("semantic view name '{}' contains null byte", name))?;
               let c_json = CString::new(json.as_str())
                   .map_err(|_| "definition JSON contains null byte".to_string())?;
               let rc = unsafe {
                   crate::shim::ffi::semantic_views_pragma_define(
                       conn,
                       c_name.as_ptr(),
                       c_json.as_ptr(),
                   )
               };
               if rc != 0 {
                   return Err(format!(
                       "failed to persist semantic view '{}': table write failed",
                       name
                   )
                   .into());
               }
           }

           // 2. Update in-memory catalog AFTER successful persist.
           //    catalog_insert validates JSON and checks for duplicates.
           catalog_insert(&state.catalog, &name, &json)?;

           let msg = format!("Semantic view '{name}' registered successfully");
           out.insert(i, msg.as_str());
       }
       ```

       IMPORTANT: The `#[cfg(feature = "extension")]` on the pragma write block ensures unit tests (default/bundled feature, no C++ shim) compile and pass. Under `cargo test`, the pragma write is skipped — `catalog_insert` alone is called.

    ### src/ddl/drop.rs

    Apply the same pattern:

    1. Remove `write_sidecar` import, keep `catalog_delete`:
       - Old: `use crate::catalog::{catalog_delete, write_sidecar, CatalogState};`
       - New: `use crate::catalog::{catalog_delete, CatalogState};`

    2. Add `use std::ffi::CString;`

    3. Replace `DropState`:
       ```rust
       /// Shared state for `drop_semantic_view`.
       /// See [`crate::ddl::define::DefineState`] for the persist_conn pattern.
       #[derive(Clone)]
       pub struct DropState {
           pub catalog: CatalogState,
           pub persist_conn: Option<libduckdb_sys::duckdb_connection>,
       }

       unsafe impl Send for DropState {}
       unsafe impl Sync for DropState {}
       ```

    4. Update `invoke` in `DropSemanticView` to write-first:

       ```rust
       for (i, raw) in names.iter().enumerate().take(input.len()) {
           let name = duckdb::types::DuckString::new(&mut { *raw })
               .as_str()
               .to_string();

           // 1. Delete from DuckDB table FIRST (write-first for consistency).
           #[cfg(feature = "extension")]
           if let Some(conn) = state.persist_conn {
               let c_name = CString::new(name.as_str())
                   .map_err(|_| format!("semantic view name '{}' contains null byte", name))?;
               let rc = unsafe {
                   crate::shim::ffi::semantic_views_pragma_drop(conn, c_name.as_ptr())
               };
               if rc != 0 {
                   return Err(format!(
                       "failed to remove semantic view '{}' from persistent storage",
                       name
                   )
                   .into());
               }
           }

           // 2. Remove from in-memory catalog AFTER successful table delete.
           catalog_delete(&state.catalog, &name)?;

           let msg = format!("Semantic view '{name}' removed successfully");
           out.insert(i, msg.as_str());
       }
       ```

    After editing both files, run `cargo test` to catch any compile errors.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test 2>&1 | tail -20</automated>
  </verify>
  <done>
    `cargo test` passes. `src/ddl/define.rs` has `persist_conn: Option<libduckdb_sys::duckdb_connection>` in `DefineState` and write-first pragma call in invoke. `src/ddl/drop.rs` has the same pattern. Neither file imports `write_sidecar` anymore.
  </done>
</task>

<task type="auto">
  <name>Task 2: Update lib.rs to create persist_conn and pass it to DefineState/DropState</name>
  <files>src/lib.rs</files>
  <action>
    Update `init_extension` in `src/lib.rs` to create a `persist_conn` for file-backed databases and pass it to `DefineState` and `DropState`.

    1. After the existing `query_conn` creation block (around line 107), add `persist_conn` creation:

       ```rust
       // Create a separate connection for DDL persistence (define_semantic_view / drop_semantic_view).
       // Only created for file-backed databases — in-memory DBs use HashMap only.
       // This connection is stored in DefineState and DropState and called from invoke
       // to execute INSERT/DELETE on semantic_layer._definitions without deadlocking
       // the main connection's execution lock.
       let persist_conn: Option<ffi::duckdb_connection> = if db_path.as_ref() != ":memory:" {
           let mut conn: ffi::duckdb_connection = ptr::null_mut();
           let rc = unsafe { ffi::duckdb_connect(db_handle, &mut conn) };
           if rc != ffi::DuckDBSuccess {
               return Err("Failed to create persist connection for DDL writes".into());
           }
           Some(conn)
       } else {
           None
       };
       ```

    2. Update `DefineState` construction (around line 80):
       ```rust
       con.register_scalar_function_with_state::<DefineSemanticView>(
           "define_semantic_view",
           &DefineState {
               catalog: catalog_state.clone(),
               persist_conn,
           },
       )?;
       ```

    3. Update `DropState` construction (around line 85):
       ```rust
       con.register_scalar_function_with_state::<DropSemanticView>(
           "drop_semantic_view",
           &DropState {
               catalog: catalog_state.clone(),
               persist_conn,
           },
       )?;
       ```

    4. The `db_path` variable is still needed for `init_catalog(con, &db_path)` — keep it. Only the references to `db_path` IN `DefineState` and `DropState` are removed. The `db_path` variable itself stays in scope for `init_catalog`.

    5. Remove the now-unused `Arc<str>` import for db_path if `db_path` is still used by `init_catalog` as `&str` — it remains in scope so nothing changes to `Arc::from`.

    After editing, run `cargo test` to confirm no compile errors.

    NOTE: The `db_path` variable is still an `Arc<str>` used in `init_catalog(con, &db_path)`. It no longer needs to be cloned into DefineState/DropState. The `Arc` clone (`db_path.clone()`) for those states is now removed. The single `Arc<str>` owned by the closure is enough.
  </action>
  <verify>
    <automated>cd /Users/paul/Documents/Dev/Personal/duckdb-semantic-views && cargo test 2>&1 | tail -20</automated>
  </verify>
  <done>
    `cargo test` passes. `grep -n "persist_conn" src/lib.rs` shows persist_conn creation and passing to DefineState/DropState. `grep -n "db_path" src/lib.rs` still shows db_path used by init_catalog but NOT passed to DefineState or DropState. No compile warnings.
  </done>
</task>

</tasks>

<verification>
Run `cargo test` — all tests pass, no warnings.

Spot checks:
- `grep -n "write_sidecar" src/ddl/define.rs src/ddl/drop.rs` — returns nothing (removed)
- `grep -n "persist_conn" src/ddl/define.rs src/ddl/drop.rs src/lib.rs` — appears in all three
- `grep -n "db_path" src/ddl/define.rs` — returns nothing (field removed)
- `grep -n "db_path" src/ddl/drop.rs` — returns nothing (field removed)
- `grep -n "pragma_define\|pragma_drop" src/ddl/define.rs src/ddl/drop.rs` — shows FFI calls
</verification>

<success_criteria>
- `cargo test` passes clean
- `DefineState` and `DropState` have `persist_conn: Option<duckdb_connection>` replacing `db_path`
- invoke write-first: pragma FFI call before catalog HashMap update
- In-memory databases: `persist_conn = None` → HashMap-only path (no FFI call)
- lib.rs creates `persist_conn` via `duckdb_connect` for file-backed databases only
- `write_sidecar` no longer called from any invoke path
</success_criteria>

<output>
After completion, create `.planning/phases/10-pragma-query-t-catalog-persistence/10-02-SUMMARY.md`
</output>
