// C++ helper for the DuckDB semantic_views extension.
//
// The Rust entry point (semantic_views_init_c_api, C_STRUCT ABI) owns the
// DuckDB handshake and function registration. After init, it calls
// sv_register_parser_hooks() here to install:
//
//   * parser_override (sv_parser_override) — the success path. Recognised
//     CREATE / DROP / ALTER / DESCRIBE / SHOW DDL is rewritten to native SQL
//     and re-parsed on the caller's connection (transactional behaviour).
//   * parse_function (sv_parse_stub) — Phase 62 Plan 03 error-reporting
//     layer. Called by DuckDB after the default parser fails on an
//     unrecognised prefix; re-runs validation and returns
//     DISPLAY_EXTENSION_ERROR with `error_location` so
//     ParserException::SyntaxError can render `LINE 1: … ^` (caret).
//   * plan_function (sv_plan_unreachable) — required sibling of
//     parse_function; should never fire because sv_parse_stub never returns
//     PARSE_SUCCESSFUL.
//
// All DuckDB C++ symbols are provided by compiling duckdb.cpp (the
// amalgamation source) alongside this file. Symbol visibility on the cdylib
// restricts exports to just the Rust entry point, so these definitions stay
// internal to the binary.
//
// DuckDB 1.5.0 moved the parser extension type declarations from duckdb.hpp
// into duckdb.cpp. The compat header re-declares them so this translation
// unit can use them. See cpp/include/parser_extension_compat.hpp for details.

#include "parser_extension_compat.hpp"
#include "shim.hpp"
#include <cstdint>
#include <cstring>
#include <memory>

using namespace duckdb;

// ---------------------------------------------------------------------------
// BORROW-contract bridge guard (Phase 65.1 Plan 10, WR-06, D-12)
// ---------------------------------------------------------------------------
// The `reinterpret_cast<duckdb_connection>(Connection*)` pattern used at ~20
// bind/exec callback sites in this file (e.g. shim.cpp:993, 1285, 1311, …)
// depends on `duckdb_connection` being layout-compatible with a single
// pointer. This compile-time guard catches DuckDB ABI drift that would
// change the size of `duckdb_connection` (e.g. wrapping it in a struct with
// extra fields). The complementary runtime probe in `sv_register_parser_hooks`
// catches representation drift the size check would miss (e.g. an indirection
// through `internal_ptr`).
//
// See cpp/src/shim.cpp BORROW contract docs and
// .planning/phases/65-overridecontext-connection-teardown/65-REVIEW.md WR-06.
static_assert(sizeof(duckdb_connection) == sizeof(void*),
    "duckdb_connection must be pointer-sized — bridge contract broken; "
    "see cpp/src/shim.cpp BORROW contract docs and 65-REVIEW.md WR-06 / "
    "Phase 65.1 D-12");

// ---------------------------------------------------------------------------
// Rust FFI declarations (defined in src/parse.rs)
// ---------------------------------------------------------------------------
extern "C" {
    // Parser-override DDL rewrite. For recognized semantic-view DDL emits
    // native SQL (INSERT/DELETE/UPDATE on _definitions for write-side DDL,
    // or `SELECT * FROM <read_side_table_function>(...)` pass-through for
    // DESCRIBE/SHOW). Returns:
    //   0 = success: heap-owned (ptr, len) written to *sql_out_ptr/*sql_out_len.
    //                Caller takes ownership and MUST release via sv_free_buffer.
    //                Buffer is NOT NUL-terminated — read exactly *sql_out_len bytes.
    //   1 = error:   message written to error_out (NUL-terminated, capped at
    //                error_out_len-1 bytes); *sql_out_ptr left untouched.
    //                (Phase 62: unused — Err branches now return rc=2 and
    //                let parse_function render caret via DISPLAY_EXTENSION_ERROR.)
    //   2 = not ours: defer to default parser.
    //
    // AR-7: the opaque Box<OverrideContext>* context parameter was removed.
    // It was empty after Phase 65 Plan 06 moved catalog pre-checks into the
    // emitted SQL, so the hook no longer needs any per-DB state.
    uint8_t sv_parser_override_rust(
        const char *query_ptr, size_t query_len,
        char **sql_out_ptr, size_t *sql_out_len,
        char *error_out, size_t error_out_len);

    // Phase 62 Plan 03: parse_function callback. Called by DuckDB after the
    // default parser fails on an unrecognised prefix. Re-runs validation
    // and returns rc + error message + byte-offset position so
    // ParserException::SyntaxError can render `LINE 1: ... ^`.
    // Return codes:
    //   0 = success/unreachable (defensive)
    //   1 = ours-but-invalid (validation error or near-miss). error_buf
    //       gets the message, position_out gets the byte offset (or
    //       UINT32_MAX if no position is available). FF-16: a panic inside
    //       the Rust hook also lands here (message
    //       "internal error: panic inside sv_parse_function_rust",
    //       position UINT32_MAX) so an extension bug surfaces as our own
    //       error instead of DuckDB's generic prefix syntax error.
    //   2 = not ours; defer (DISPLAY_ORIGINAL_ERROR on the C++ side).
    //   3 = valid DDL but parser_override didn't fire (e.g. override
    //       setting reset by disable_peg_parser). error_buf gets an
    //       actionable hint; position_out gets 0.
    uint8_t sv_parse_function_rust(
        const char *query_ptr, size_t query_len,
        char *error_buf, size_t error_buf_len,
        uint32_t *position_out);

    // Releases a buffer previously produced by sv_parser_override_rust.
    // Safe to call with a null pointer (no-op). ptr/len must be the exact
    // pair the Rust side returned.
    void sv_free_buffer(char *ptr, size_t len);

    // Phase 65 Plan 04 (Task 2 Step C) — Rust FFI bridge for the
    // `__sv_compute_create_from_yaml` helper TF. The bind callback reads the
    // file content directly through DuckDB's FileSystem API (Phase 65.1
    // D-01..D-03 replaced the earlier read_text(?) SQL path) and this bridge
    // parses + enriches the YAML on the Rust side, returning the
    // metadata-less JSON definition as a heap-owned (ptr, len) buffer.
    //
    // The outer parser_override INSERT wraps `new_def` with json_merge_patch
    // to add now()/current_database()/current_schema() on the caller's
    // connection (matching the metadata-via-SQL pattern landed in Plan 03
    // for the inline CREATE path).
    //
    // Parameters:
    //   content_ptr/len — YAML bytes loaded by the C++ bind callback.
    //   name_ptr/len    — view name (bare identifier).
    //   comment_ptr/len — optional COMMENT='...' value; pass len=0 for
    //                     "no comment provided" (the helper leaves
    //                     `def.comment` untouched in that case so the
    //                     YAML's own `comment:` field, if any, survives).
    //   out_ptr/out_len — on rc=0, point to a heap-owned UTF-8 buffer
    //                     (NOT NUL-terminated) holding the JSON definition.
    //                     Caller MUST release via `sv_free_buffer` with
    //                     the exact (ptr, len) pair Rust returned.
    //   error_buf       — on rc!=0, gets a NUL-terminated error message
    //                     capped at `error_buf_len-1` bytes. Untouched
    //                     on success.
    //
    // Return codes:
    //   0 — success; (out_ptr, out_len) populated.
    //   1 — YAML parse / size-cap error; error_buf populated.
    //   2 — enrichment / validation error; error_buf populated.
    //   3 — internal error (panic across FFI, allocation failure, etc.);
    //       error_buf populated.
    // Phase 65.1 Plan 07 (IN-04 D-24): `kind` parameter removed — the outer
    // parser_override INSERT shape (OR IGNORE / OR REPLACE / plain) already
    // encodes ON CONFLICT behaviour, so the helper itself does not branch on
    // it. Three-arg helper TF signature matches: (file_path, view_name,
    // comment).
    uint8_t sv_compute_create_from_yaml_rust(
        const uint8_t *content_ptr, size_t content_len,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *comment_ptr, size_t comment_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);

    // Phase 65 Plan 05 (Task 1 / Wave 0 bridge spike) — Rust dispatcher for
    // `list_semantic_views()`. The C++ bind callback opens a per-call
    // `Connection probe(*context.db)` and bridges by casting the stack
    // `Connection *` to `duckdb_connection` (cast confirmed by
    // duckdb.cpp:266432-266447 where `duckdb_connect` is literally
    // `reinterpret_cast<duckdb_connection>(new Connection(...))`, so the
    // C-API handle is a borrowed pointer to a C++ `Connection`).
    //
    // The bridge is a BORROW, not a transfer: the Rust dispatcher does NOT
    // call `duckdb_disconnect` (which would `delete` the stack Connection
    // and segfault). When the C++ bind scope ends, the `probe` local's
    // destructor runs and `~Connection()` does the correct teardown.
    //
    // The dispatcher serializes the result rows into a length-prefixed
    // binary buffer (`u32 row_count; for each row: for each of 6 cols: u32
    // byte_len; bytes...`) which the C++ bind parses into BindData. The
    // exec callback then emits rows from BindData. Using a flat binary
    // wire format avoids needing matched struct layouts across the FFI
    // boundary.
    //
    // Return codes:
    //   0 — success; (out_ptr, out_len) populated with the binary buffer.
    //        Caller MUST release via `sv_free_buffer`.
    //   1 — catalog read error; error_buf populated.
    //   2 — internal error (panic across FFI); error_buf populated.
    uint8_t sv_list_semantic_views_bind_rust(
        duckdb_connection conn,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);

    // Phase 65 Plan 05 Task 2 (Wave 1) — Rust dispatcher for the migrated
    // `list_terse_semantic_views()` table function. 5-column subset of
    // list_semantic_views (no `comment`). Same bridge mechanism and borrow
    // contract as the Wave 0 spike.
    uint8_t sv_list_terse_semantic_views_bind_rust(
        duckdb_connection conn,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);

    // Phase 65 Plan 05 Task 2 (Wave 1) — Rust dispatchers for the migrated
    // zero-arg "_all" TFs. All emit homogeneous VARCHAR rows; cell layout
    // matches the matching legacy duckdb-rs VTab. See per-dispatcher Rust
    // doc-headers for the exact column order.
    uint8_t sv_show_semantic_dimensions_all_bind_rust(
        duckdb_connection conn,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_metrics_all_bind_rust(
        duckdb_connection conn,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_facts_all_bind_rust(
        duckdb_connection conn,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_materializations_all_bind_rust(
        duckdb_connection conn,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);

    // Phase 65 Plan 05 Task 3 (Wave 2) — Rust dispatchers for the migrated
    // single-arg (view name) and two-arg (view name + metric name) TFs.
    // All take a per-call borrowed duckdb_connection plus name (ptr, len)
    // tuples. The single-view show_columns / describe / show_semantic_*
    // variants emit VARCHAR rows; show_semantic_dimensions_for_metric
    // emits VARCHAR+BOOL rows (3 VARCHAR + 1 trailing BOOL per row).
    uint8_t sv_show_columns_in_semantic_view_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_describe_semantic_view_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_dimensions_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_metrics_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_facts_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_materializations_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_show_semantic_dimensions_for_metric_bind_rust(
        duckdb_connection conn,
        const uint8_t *view_name_ptr, size_t view_name_len,
        const uint8_t *metric_name_ptr, size_t metric_name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);

    // Phase 65 Plan 05 Task 4 (Wave 3) — Rust dispatchers for the migrated
    // scalar functions (get_ddl, read_yaml_from_semantic_view). Per-row
    // dispatch: each invocation takes the borrowed per-call duckdb_connection
    // (opened from the C++ exec callback via Connection probe(*state.GetContext().db))
    // plus the input string args, and returns a heap-owned UTF-8 buffer for
    // the C++ side to copy into the result Vector via StringVector::AddString.
    //
    // Same wire convention as the bind dispatchers: rc=0 on success with
    // (out_ptr, out_len) populated (caller frees via sv_free_buffer), rc=1
    // on user-visible error (error_buf populated, raised as
    // InvalidInputException by the C++ side), rc=2 on internal panic.
    // `qualify` is the optional third GET_DDL argument (0 = bare name, non-zero
    // = schema-qualified). The two-argument arity passes 0.
    uint8_t sv_get_ddl_exec_rust(
        duckdb_connection conn,
        const uint8_t *type_ptr, size_t type_len,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        uint8_t qualify,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
    uint8_t sv_read_yaml_from_semantic_view_exec_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);

    // Phase 65 Plan 05 Task 5 (Wave 5) — Rust dispatcher for the migrated
    // `explain_semantic_view(view_name, dimensions := [...], metrics := [...],
    // facts := [...])` table function. Same per-call Connection BORROW
    // contract as the 14 Batch-1 migrations. The three optional named
    // LIST(VARCHAR) parameters are flattened on the C++ side into the
    // standard length-prefixed wire format (`u32 count; for each entry:
    // u32 len + bytes`) and passed as (ptr, len) pairs. A null pointer
    // with len=0 means the named parameter was not supplied (treated as
    // an empty list).
    uint8_t sv_explain_semantic_view_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        const uint8_t *dims_ptr, size_t dims_len,
        const uint8_t *metrics_ptr, size_t metrics_len,
        const uint8_t *facts_ptr, size_t facts_len,
        const uint8_t *where_ptr, size_t where_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);

    // Phase 65 Plan 05 Task 6 (Wave 6) — Rust dispatcher for the bind half
    // of the migrated `semantic_view(view_name, dimensions := [...],
    // metrics := [...], facts := [...])` table function. Same per-call
    // BORROW contract as Wave 5. Performs catalog lookup + expand + LIMIT-0
    // type-inference + execution-SQL construction, and returns:
    //
    //   wire format (u32 little-endian unless noted):
    //     u32 n_cols
    //     for each column:
    //       u32 byte_len + bytes (column name, UTF-8)
    //       u32 duckdb_type_id (normalised: HUGEINT→BIGINT, UHUGEINT→UBIGINT,
    //                           STRUCT/MAP/INVALID→VARCHAR for declaration)
    //     u32 byte_len + bytes (execution_sql, UTF-8)
    //
    // The C++ bind callback parses the buffer, declares the output schema
    // (handling DECIMAL/LIST/ENUM logical-type metadata via a second
    // LIMIT-0 query on the per-call Connection only when needed), stashes
    // the execution_sql in BindData, and runs the actual query inside
    // init_global so chunks can be streamed during exec.
    uint8_t sv_semantic_view_bind_rust(
        duckdb_connection conn,
        const uint8_t *name_ptr, size_t name_len,
        const uint8_t *sp_ptr, size_t sp_len,
        const uint8_t *dims_ptr, size_t dims_len,
        const uint8_t *metrics_ptr, size_t metrics_len,
        const uint8_t *facts_ptr, size_t facts_len,
        const uint8_t *where_ptr, size_t where_len,
        char **out_ptr, size_t *out_len,
        char *error_buf, size_t error_buf_len);
}

// ---------------------------------------------------------------------------
// SemanticViewParseData — Phase 62 Plan 03
// ---------------------------------------------------------------------------
// Concrete subclass of ParserExtensionParseData required by the
// ParserExtensionParseResult(unique_ptr<...>) constructor. We never return
// PARSE_SUCCESSFUL from sv_parse_stub (its sole purpose is rendering errors
// via DISPLAY_EXTENSION_ERROR), so this type is structurally needed only
// for layout / type-system reasons. If we ever do produce a parse_data,
// sv_plan_unreachable below would fire.
struct SemanticViewParseData : public ParserExtensionParseData {
    string query;
    explicit SemanticViewParseData(string q) : query(std::move(q)) {}

    unique_ptr<ParserExtensionParseData> Copy() const override {
        return make_uniq<SemanticViewParseData>(query);
    }
    string ToString() const override {
        return query;
    }
};

// RAII guard for heap-owned buffers returned by the Rust FFI. Ensures the
// buffer is released even if a downstream call (Parser::ParseQuery) throws.
struct SvOwnedBuffer {
    char *ptr = nullptr;
    size_t len = 0;
    SvOwnedBuffer() = default;
    SvOwnedBuffer(const SvOwnedBuffer &) = delete;
    SvOwnedBuffer &operator=(const SvOwnedBuffer &) = delete;
    SvOwnedBuffer(SvOwnedBuffer &&other) noexcept
        : ptr(other.ptr), len(other.len) {
        other.ptr = nullptr;
        other.len = 0;
    }
    SvOwnedBuffer &operator=(SvOwnedBuffer &&other) noexcept {
        if (this != &other) {
            if (ptr) sv_free_buffer(ptr, len);
            ptr = other.ptr;
            len = other.len;
            other.ptr = nullptr;
            other.len = 0;
        }
        return *this;
    }
    ~SvOwnedBuffer() {
        if (ptr) sv_free_buffer(ptr, len);
    }
    string to_string() const {
        return ptr ? string(ptr, len) : string();
    }
};

// Per-extension-load marker attached to ParserExtension::parser_info. AR-7:
// this used to carry an opaque Box<OverrideContext>* (rust_state), but the
// OverrideContext was empty after Phase 65 Plan 06's H1 catalog_conn
// retirement, so it now holds no state. It is kept purely as the
// `dynamic_cast<SemanticViewsParserInfo *>` marker type that `sv_parser_override`
// uses to confirm the parser_info is ours. (`sv_parse_stub` no longer consults
// parser_info at all — validation needs no per-DB state — so it ignores the
// marker.)
struct SemanticViewsParserInfo : public ParserExtensionInfo {};

// ---------------------------------------------------------------------------
// Parser-override hook: sv_parser_override
// ---------------------------------------------------------------------------
// The sole DDL entry point. Runs *before* the default parser. Recognized
// semantic-view DDL is rewritten into native SQL by the Rust side and
// re-parsed via DuckDB's own Parser, producing SQLStatement ASTs that DuckDB
// then plans and executes on the caller's connection — so write-side DDL
// participates in the caller's transaction.
//
// For non-matching queries returns DISPLAY_ORIGINAL_ERROR so DuckDB falls
// through to the default parser.
static ParserOverrideResult sv_parser_override(
    ParserExtensionInfo *info, const string &query, ParserOptions &) {

    // Confirm this is our parser_info. info is the per-extension-load
    // SemanticViewsParserInfo attached at registration time; if it isn't ours
    // (dynamic_cast fails) defer to the default parser. AR-7: there is no
    // longer any per-DB Rust state to route through.
    auto *sv_info = dynamic_cast<SemanticViewsParserInfo *>(info);
    if (!sv_info) {
        return ParserOverrideResult();
    }

    SvOwnedBuffer sql_buf;
    char error_buf[1024];
    memset(error_buf, 0, sizeof(error_buf));

    uint8_t rc = sv_parser_override_rust(
        query.c_str(), query.size(),
        &sql_buf.ptr, &sql_buf.len,
        error_buf, sizeof(error_buf));

    if (rc == 2) {
        // Not our query — let DuckDB's default parser handle it.
        return ParserOverrideResult();
    }

    if (rc == 1) {
        // Phase 62 Plan 03: defer to default parser. parse_function
        // (registered as sv_parse_stub) re-runs validation and returns
        // DISPLAY_EXTENSION_ERROR with caret position. The Rust side
        // currently always returns rc=2 on the error path, so this
        // branch is defensive — kept to match the documented contract.
        return ParserOverrideResult();
    }

    // rc == 0: native SQL produced. Re-parse via DuckDB's Parser. Use
    // default-constructed ParserOptions so parser_override doesn't recurse
    // (DEFAULT_OVERRIDE skips parser_override hooks entirely). The
    // rewritten SQL is read by exact length, so size is unbounded.
    string native_sql = sql_buf.to_string();
    try {
        Parser parser;
        parser.ParseQuery(native_sql);
        return ParserOverrideResult(std::move(parser.statements));
    } catch (std::exception &e) {
        return ParserOverrideResult(e);
    }
}

// ---------------------------------------------------------------------------
// Parse-function hook: sv_parse_stub  (Phase 62 Plan 03)
// ---------------------------------------------------------------------------
// Called by DuckDB's Parser::ParseQuery after the default parser fails on
// an unrecognised prefix. Sole purpose is rendering caret-aware errors via
// ParserException::SyntaxError(query, msg, error_location).
//
// parser_override remains the success path — it rewrites recognized DDL
// to native SQL, re-parses on the caller's connection, and returns
// PARSE_SUCCESSFUL (transactional behaviour preserved). For error cases
// parser_override now ALWAYS returns DISPLAY_ORIGINAL_ERROR (rc=2 from
// Rust), letting the default parser fail and DuckDB call this stub.
//
// We re-run validation through sv_parse_function_rust which returns:
//   0 — defensive internal error; render as DISPLAY_EXTENSION_ERROR
//   1 — ours-but-invalid: error_buf populated; position is byte offset.
//       FF-16: also covers a panic inside the Rust hook (position
//       UINT32_MAX) — an extension bug must not masquerade as the user's
//       syntax error, which is what rc=2 on panic used to produce.
//   2 — not ours: DISPLAY_ORIGINAL_ERROR (let the default parser's error stand)
//   3 — valid-but-override-disabled: actionable hint; position=0
static ParserExtensionParseResult sv_parse_stub(
    ParserExtensionInfo * /*info*/, const string &query) {
    // AR-7: sv_parse_function_rust no longer takes a context pointer — the
    // validation path never needed the catalog — so the SemanticViewsParserInfo
    // marker is not consulted here.
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));
    uint32_t position = UINT32_MAX;

    uint8_t rc = sv_parse_function_rust(
        query.c_str(), query.size(),
        error_buf, sizeof(error_buf),
        &position);

    switch (rc) {
        case 2:
            // Not ours — DISPLAY_ORIGINAL_ERROR (default parser's error
            // text + caret stands).
            return ParserExtensionParseResult();

        case 1:
        case 3: {
            // ours-but-invalid OR valid-but-override-disabled. Both want
            // DISPLAY_EXTENSION_ERROR with caret position, just different
            // message text. Construct the std::string explicitly first to
            // dodge C++'s "most vexing parse" — `ParserExtensionParseResult
            // result(string(error_buf));` would otherwise be parsed as a
            // function declaration.
            string msg(error_buf);
            ParserExtensionParseResult result(std::move(msg));
            if (position != UINT32_MAX) {
                result.error_location = optional_idx(position);
            }
            return result;
        }

        case 0: {
            // Defensive — sv_parse_function_rust currently never returns 0.
            // Map to DISPLAY_EXTENSION_ERROR with an internal-error message
            // rather than letting a silent default-parser error escape.
            return ParserExtensionParseResult(string(
                "semantic_views: internal error — sv_parse_function_rust "
                "returned rc=0 (please report this bug)"));
        }

        default: {
            // Defensive — unknown rc. Same handling as rc=0.
            return ParserExtensionParseResult(string(
                "semantic_views: internal error — unknown rc from "
                "sv_parse_function_rust"));
        }
    }
}

// ---------------------------------------------------------------------------
// Plan-function hook: sv_plan_unreachable  (Phase 62 Plan 03)
// ---------------------------------------------------------------------------
// ParserExtension carries a sibling `plan_function` pointer alongside
// `parse_function`. The plan function only fires when parse_function returns
// PARSE_SUCCESSFUL, which sv_parse_stub never does — every code path goes
// through DISPLAY_EXTENSION_ERROR or DISPLAY_ORIGINAL_ERROR. Provide a
// hard-fail stub so a contract violation surfaces loudly rather than silently.
//
// Signature must match plan_function_t in parser_extension_compat.hpp:
//   ParserExtensionPlanResult (*)(ParserExtensionInfo *, ClientContext &,
//                                  unique_ptr<ParserExtensionParseData>);
static ParserExtensionPlanResult sv_plan_unreachable(
    ParserExtensionInfo * /*info*/,
    ClientContext & /*context*/,
    unique_ptr<ParserExtensionParseData> /*parse_data*/) {
    throw InternalException(
        "semantic_views: sv_plan_unreachable called — sv_parse_stub never "
        "returns PARSE_SUCCESSFUL (please report this bug)");
}

// ---------------------------------------------------------------------------
// Table-function registration: shared spec-based core (C-7, code-review
// 2026-07-11)
// ---------------------------------------------------------------------------
// Every table function this extension installs goes through the C++ Catalog
// API with identical boilerplate: unwrap the `DatabaseWrapper`, build a
// `TableFunction`, set `ALTER_ON_CONFLICT`, and register it on the
// system-catalog transaction — all inside a try/catch that funnels failures
// into `error_buf` (D-08/D-09, never stderr). That boilerplate used to live in
// three places: the generic `sv_register_table_function` wrapper plus two
// hand-rolled impls for the named-parameter TFs (`explain_semantic_view`,
// `semantic_view`) that the generic wrapper could not express. `SvTableFunctionSpec`
// + `sv_register_table_function_core` own it once; the wrapper and the two named
// registrations are now thin spec builders that differ only in data.
struct SvTableFunctionSpec {
    const char *name = nullptr;
    const LogicalType *arg_types = nullptr;
    size_t arg_count = 0;
    // Named parameters, applied in order (empty for positional-only TFs).
    std::vector<std::pair<std::string, LogicalType>> named_params;
    table_function_bind_t bind_cb = nullptr;
    table_function_t exec_cb = nullptr;
    // D-05 (generalized): exactly one per-execution state constructor. A TF
    // uses init_local (single-shot / streaming via local state) XOR init_global
    // (semantic_view's per-execution global). The core enforces the XOR.
    table_function_init_local_t init_local_cb = nullptr;
    table_function_init_global_t init_global_cb = nullptr;
};

// The named LIST(VARCHAR) parameters shared by explain_semantic_view and
// semantic_view. Defined once (C-7) so the two TFs cannot drift from each
// other: this triple must stay byte-for-byte identical to the legacy Rust VTab
// signature (`dimensions`, `metrics`, `facts`) so existing call sites keep
// parsing.
static std::vector<std::pair<std::string, LogicalType>> sv_semantic_named_params() {
    auto list_varchar = LogicalType::LIST(LogicalType::VARCHAR);
    return {
        {"dimensions", list_varchar},
        {"metrics", list_varchar},
        {"facts", list_varchar},
        // The pre-aggregation predicate — Snowflake's `SEMANTIC_VIEW( … WHERE
        // <predicate> )`. Named `where_clause` rather than `where` because
        // DuckDB's parser reserves that keyword in named-parameter position:
        // `where := '…'` is a syntax error before the binder ever sees it.
        {"where_clause", LogicalType::VARCHAR},
        // The caller's schema resolution order, injected by the parser
        // override (see `crate::parse::search_path::SEARCH_PATH_SQL`). The read side binds
        // on a fresh connection that cannot see the caller's search path or
        // current schema, so the only way it can resolve an unqualified view
        // name the way DuckDB would is to be handed the path — TECH-DEBT #19.
        {"search_path", list_varchar},
    };
}

// Shared registration core. `caller` names the outer entry point so
// diagnostics keep naming the specific function (e.g. "sv_register_semantic_view
// failed: ..."). Every failure path writes a NUL-terminated diagnostic into
// `error_buf` (never stderr — D-09) and returns false.
static bool sv_register_table_function_core(
    duckdb_database db_handle,
    const SvTableFunctionSpec &spec,
    const char *caller,
    char *error_buf, size_t error_buf_len) {
    auto write_err = [error_buf, error_buf_len](const std::string &msg) {
        if (error_buf == nullptr || error_buf_len == 0) {
            return;
        }
        snprintf(error_buf, error_buf_len, "%s", msg.c_str());
    };
    // Diagnostics name both the registration entry point (`caller`) and the
    // specific table function (`spec.name`), reproducing the pre-C-7
    // `sv_register_table_function('<tf_name>') ...` format so no failure path
    // — including the wrapper's — loses the TF name.
    const std::string who =
        std::string(caller != nullptr ? caller : "sv_register_table_function_core") +
        "('" + (spec.name != nullptr ? spec.name : "(null)") + "')";
    try {
        if (db_handle == nullptr || spec.name == nullptr ||
            spec.bind_cb == nullptr || spec.exec_cb == nullptr) {
            write_err(who + ": null required argument");
            return false;
        }
        // D-05 generalized (CR-02): every TF must have exactly one
        // per-execution state constructor so the single-shot/streaming exec
        // cannot double-emit or loop. init_local XOR init_global — both-null
        // reopens that hazard; both-set mixes the two state models.
        const bool has_local = spec.init_local_cb != nullptr;
        const bool has_global = spec.init_global_cb != nullptr;
        if (has_local == has_global) {
            write_err(who +
                ": exactly one of init_local / init_global must be non-null "
                "(D-05)");
            return false;
        }
        // Symmetric null guard for arg_types when arg_count > 0 (WR-01): a
        // buggy (nullptr, > 0) caller would otherwise segfault in the copy loop.
        if (spec.arg_count > 0 && spec.arg_types == nullptr) {
            write_err(who + ": arg_count > 0 but arg_types is null");
            return false;
        }
        auto *wrapper = reinterpret_cast<duckdb::DatabaseWrapper *>(
            db_handle->internal_ptr);
        if (wrapper == nullptr) {
            write_err(who + ": null DatabaseWrapper");
            return false;
        }
        auto &db = *wrapper->database->instance;

        vector<LogicalType> args;
        args.reserve(spec.arg_count);
        for (size_t i = 0; i < spec.arg_count; ++i) {
            args.push_back(spec.arg_types[i]);
        }

        // Six-arg TableFunction ctor: (name, args, function, bind,
        // init_global, init_local). Exactly one init callback is non-null
        // (enforced above); the other is nullptr.
        TableFunction tf(
            std::string(spec.name),
            std::move(args),
            spec.exec_cb,
            spec.bind_cb,
            spec.init_global_cb,
            spec.init_local_cb);

        for (const auto &np : spec.named_params) {
            tf.named_parameters[np.first] = np.second;
        }

        CreateTableFunctionInfo info(tf);
        // ALTER_ON_CONFLICT: extension reload replaces the registration cleanly
        // instead of throwing on duplicate name.
        info.on_conflict = OnCreateConflict::ALTER_ON_CONFLICT;

        auto &system_catalog = Catalog::GetSystemCatalog(db);
        auto txn = CatalogTransaction::GetSystemTransaction(db);
        system_catalog.CreateTableFunction(txn, info);
        return true;
    } catch (const std::exception &e) {
        write_err(who + " failed: " + e.what());
        return false;
    } catch (...) {
        write_err(who + " failed: unknown C++ exception");
        return false;
    }
}

// ---------------------------------------------------------------------------
// sv_register_table_function — Phase 65 Plan 04 (A2 resolution)
// ---------------------------------------------------------------------------
// Reusable C-callable wrapper around the C++ Catalog API table-function
// registration pattern proven by `65-READ-PATH-SPIKE.md`. Bind callbacks
// registered via this path receive a native `ClientContext &` (not the
// duckdb-rs `BindInfo` wrapper which marshals `ClientContext` away), so
// they can open per-call `Connection(*context.db)` for catalog reads and
// YAML parsing without needing a long-lived extension-owned connection.
//
// Plan 04 consumes this to register `__sv_compute_create_from_yaml`. Plan 05
// will consume it (and add a scalar sibling) to migrate the 17 read-side
// callbacks off `query_conn` (H2).
//
// Header declaration in cpp/src/shim.hpp.
// Phase 65.1 Plan 02a (WR-02 D-08/D-09 + CR-02 D-05) — registration
// failures surface via the `(error_buf, error_buf_len)` trailing pair,
// the same ABI-stable channel used by `sv_parser_override_rust`
// (shim.cpp:57-61). Per D-09 there is NO stderr write — ADBC/JDBC/Python
// callers may have redirected stderr; `error_buf` is the only reliable
// path. D-05: `init_cb == nullptr` is rejected at registration time
// (forces every TF onto the single-shot-via-local-state path so the
// double-emit / unbounded-loop hazard in CR-02 cannot recur).
extern "C" {
    bool sv_register_table_function(
        duckdb_database db_handle,
        const char *name,
        const LogicalType *arg_types,
        size_t arg_count,
        table_function_bind_t bind_cb,
        table_function_t exec_cb,
        table_function_init_local_t init_cb,
        char *error_buf, size_t error_buf_len) {
        // Thin compatibility wrapper (C-7): preserves this entry point's
        // narrow, documented contract (shim.hpp) — its call sites are all
        // single-shot local-state TFs, so init_cb (init_local) is MANDATORY
        // here and the exact legacy diagnostic is kept. Construction, the
        // generalized D-05 XOR check, named-parameter population, and the
        // catalog transaction all live in the shared core above.
        if (init_cb == nullptr) {
            if (error_buf != nullptr && error_buf_len > 0) {
                snprintf(error_buf, error_buf_len,
                    "sv_register_table_function('%s'): null required "
                    "argument (init_cb is mandatory)",
                    name ? name : "(null)");
            }
            return false;
        }
        SvTableFunctionSpec spec;
        spec.name = name;
        spec.arg_types = arg_types;
        spec.arg_count = arg_count;
        spec.bind_cb = bind_cb;
        spec.exec_cb = exec_cb;
        spec.init_local_cb = init_cb;
        spec.init_global_cb = nullptr;
        // Every read TF registered through this wrapper accepts the caller's
        // schema resolution order. Declared uniformly rather than per-TF: the
        // parser override injects it into any read-TF call it rewrites, and a
        // TF that takes no view name simply ignores it. `semantic_view` and
        // `explain_semantic_view` declare it via `sv_semantic_named_params`.
        spec.named_params.emplace_back("search_path",
                                       LogicalType::LIST(LogicalType::VARCHAR));
        return sv_register_table_function_core(
            db_handle, spec, "sv_register_table_function", error_buf,
            error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// sv_register_scalar_function — Phase 65 Plan 05 (Task 2 Step A)
// ---------------------------------------------------------------------------
// Sibling of `sv_register_table_function`. Registers a scalar function via
// the C++ Catalog API (`Catalog::CreateFunction` on the system catalog with
// a `CreateScalarFunctionInfo`). Consumed by Task 4 / Wave 4 to migrate
// `get_ddl` and `read_yaml_from_semantic_view` off
// `register_scalar_function_with_state`.
//
// Header declaration in cpp/src/shim.hpp.
// Phase 65.1 Plan 02a (WR-02 D-08/D-09) — see sv_register_table_function
// for the rationale. Scalar functions have no `init_local` concept, so
// D-05 (null-init refusal) does NOT apply here; the required-arg set
// stays db_handle / name / exec_cb.
//
// TECH-DEBT #25 (GET_DDL qualified names): the catalog write is factored into
// `sv_register_scalar_function_set` so a function carrying more than one arity
// — `get_ddl`, whose third argument is optional — registers through the same
// guards and error-buffer contract as the single-arity callers. The C entry
// point below keeps its signature and wraps its one overload into a set.
static bool sv_register_scalar_function_set(
    duckdb_database db_handle,
    const char *name,
    ScalarFunctionSet set,
    char *error_buf, size_t error_buf_len) {
    auto write_err = [error_buf, error_buf_len](const char *fmt,
                                                const char *arg1) {
        if (error_buf == nullptr || error_buf_len == 0) {
            return;
        }
        snprintf(error_buf, error_buf_len, fmt, arg1 ? arg1 : "(null)");
    };
    try {
        if (db_handle == nullptr || name == nullptr) {
            write_err(
                "sv_register_scalar_function('%s'): null required argument",
                name);
            return false;
        }
        auto *wrapper = reinterpret_cast<duckdb::DatabaseWrapper *>(
            db_handle->internal_ptr);
        if (wrapper == nullptr) {
            write_err(
                "sv_register_scalar_function('%s'): null DatabaseWrapper",
                name);
            return false;
        }
        auto &db = *wrapper->database->instance;

        CreateScalarFunctionInfo info(std::move(set));
        info.on_conflict = OnCreateConflict::ALTER_ON_CONFLICT;

        auto &system_catalog = Catalog::GetSystemCatalog(db);
        auto txn = CatalogTransaction::GetSystemTransaction(db);
        system_catalog.CreateFunction(txn, info);
        return true;
    } catch (const std::exception &e) {
        if (error_buf != nullptr && error_buf_len > 0) {
            snprintf(error_buf, error_buf_len,
                "sv_register_scalar_function('%s') failed: %s",
                name ? name : "(null)", e.what());
        }
        return false;
    } catch (...) {
        write_err(
            "sv_register_scalar_function('%s') failed: unknown C++ exception",
            name);
        return false;
    }
}

extern "C" {
    bool sv_register_scalar_function(
        duckdb_database db_handle,
        const char *name,
        const LogicalType *arg_types,
        size_t arg_count,
        LogicalType return_type,
        scalar_function_t exec_cb,
        char *error_buf, size_t error_buf_len) {
        auto write_err = [error_buf, error_buf_len](const char *fmt,
                                                    const char *arg1) {
            if (error_buf == nullptr || error_buf_len == 0) {
                return;
            }
            snprintf(error_buf, error_buf_len, fmt, arg1 ? arg1 : "(null)");
        };
        try {
            if (name == nullptr || exec_cb == nullptr) {
                write_err(
                    "sv_register_scalar_function('%s'): null required argument",
                    name);
                return false;
            }
            // Phase 65.1 WR-01: symmetric null guard for arg_types when the
            // caller declares a non-zero arg_count. Same contract as the
            // table-function sibling above.
            if (arg_count > 0 && arg_types == nullptr) {
                write_err(
                    "sv_register_scalar_function('%s'): arg_count > 0 but "
                    "arg_types is null",
                    name);
                return false;
            }

            vector<LogicalType> args;
            args.reserve(arg_count);
            for (size_t i = 0; i < arg_count; ++i) {
                args.push_back(arg_types[i]);
            }

            // Braces, not parens: `ScalarFunctionSet set(std::string(name));`
            // is a function declaration (most vexing parse), not a variable.
            ScalarFunctionSet set{std::string(name)};
            set.AddFunction(ScalarFunction(std::string(name), std::move(args),
                                           return_type, exec_cb));
            return sv_register_scalar_function_set(
                db_handle, name, std::move(set), error_buf, error_buf_len);
        } catch (const std::exception &e) {
            if (error_buf != nullptr && error_buf_len > 0) {
                snprintf(error_buf, error_buf_len,
                    "sv_register_scalar_function('%s') failed: %s",
                    name ? name : "(null)", e.what());
            }
            return false;
        } catch (...) {
            write_err(
                "sv_register_scalar_function('%s') failed: unknown C++ exception",
                name);
            return false;
        }
    }
}

// ---------------------------------------------------------------------------
// __sv_compute_create_from_yaml — Phase 65 Plan 04 (Task 2 Step B) +
//                                  Phase 65.1 Plan 07 (CR-01 D-01..D-03)
// ---------------------------------------------------------------------------
// Helper table function registered via `sv_register_table_function`. The bind
// callback reads the YAML file directly through DuckDB's `FileSystem` API
// (`FileSystem::GetFileSystem(context).OpenFile(path, FileFlags::FILE_FLAGS_READ)`
// → `GetFileSize` → `Read`). It then calls the Rust FFI helper
// `sv_compute_create_from_yaml_rust` to parse + enrich + serialize the YAML
// into a metadata-less JSON definition. The exec callback emits the JSON
// as a single VARCHAR row.
//
// Phase 65.1 D-01..D-03 (CR-01): the previous shape used
// `Connection probe(*context.db)` + `Query("SELECT content FROM
// read_text('<path>')")`, which required SQL-quote-doubling on the path. That
// surface is gone: the file is now read directly via the `FileSystem` API.
// `LocalFileSystem` continues to honour the global `enable_external_access`
// DBConfig option when no `FileOpener` is supplied, so the access gate is
// preserved by construction. The four `BinderException` paths below all
// retain the `"FROM YAML FILE failed: "` prefix pinned by
// `test/sql/phase53_yaml_file.test`.
//
// The outer parser_override INSERT (src/parse.rs::rewrite_yaml_file_create,
// landed in Task 4) wraps the helper TF's row via `json_merge_patch(new_def,
// json_object('created_on', strftime(now(), '%Y-%m-%dT%H:%M:%SZ'),
// 'database_name', current_database(), 'schema_name', current_schema()))`
// so metadata fields are populated on the caller's connection, preserving
// the D-21 transactional contract.

struct CreateFromYamlBindData : public TableFunctionData {
    std::string file_path;
    std::string view_name;
    std::string comment;
    std::string new_def;
};

struct CreateFromYamlLocalState : public LocalTableFunctionState {
    bool emitted = false;
};

static unique_ptr<FunctionData> sv_create_from_yaml_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<CreateFromYamlBindData>();
    // FF-15 (code-review 2026-08-08): guard every positional argument before
    // GetValue<string>(), the same up-front check FF-4 gave every other table
    // function (see sv_run_varchar_bind_with_name / sv_semantic_view_bind).
    // `Value::GetValue<string>()` on a NULL Value renders the literal text
    // "NULL", so without this a NULL argument silently became a file named
    // `NULL` / a view named `NULL` instead of an error.
    //
    // The rewriter (src/parse/native_sql.rs::rewrite_yaml_file_create) always
    // emits string literals here, so this is only reachable by calling the
    // helper TF directly — hence the function-name-prefixed FF-4 wording
    // rather than the `"FROM YAML FILE failed: "` prefix the file-read paths
    // below use (that prefix belongs to the user-facing DDL surface and is
    // pinned by `test/sql/phase53_yaml_file.test`).
    // Exactly 3, not "at least 3": the message promises a fixed signature, and
    // a 4-argument call silently ignoring the extra is the kind of quiet
    // tolerance that makes a helper's contract untrue. DuckDB's own arity check
    // should reject it first, so this is defence in depth either way.
    if (input.inputs.size() != 3) {
        throw BinderException(
            "__sv_compute_create_from_yaml: expected 3 arguments "
            "(file_path, view_name, comment)");
    }
    if (input.inputs[0].IsNull()) {
        throw BinderException(
            "__sv_compute_create_from_yaml: file path is required "
            "(positional arg 0)");
    }
    if (input.inputs[1].IsNull()) {
        throw BinderException(
            "__sv_compute_create_from_yaml: view name is required "
            "(positional arg 1)");
    }
    if (input.inputs[2].IsNull()) {
        throw BinderException(
            "__sv_compute_create_from_yaml: comment is required "
            "(positional arg 2; pass '' for no comment)");
    }
    bd->file_path = input.inputs[0].GetValue<string>();
    bd->view_name = input.inputs[1].GetValue<string>();
    bd->comment   = input.inputs[2].GetValue<string>();

    // Phase 65.1 D-01..D-03 (CR-01): read the YAML file directly via the
    // `FileSystem` API rather than `Connection::Query("SELECT content FROM
    // read_text('<path>')")`. The SQL surface — including its quote-doubling
    // escape — is gone; `LocalFileSystem` honours `enable_external_access`
    // natively when no `FileOpener` is supplied, so the access gate is
    // preserved by construction. Every throw below keeps the
    // `"FROM YAML FILE failed: "` prefix pinned by
    // `test/sql/phase53_yaml_file.test`.
    auto &fs = FileSystem::GetFileSystem(context);
    unique_ptr<FileHandle> handle;
    try {
        handle = fs.OpenFile(bd->file_path, FileFlags::FILE_FLAGS_READ);
    } catch (const std::exception &e) {
        throw BinderException(
            std::string("FROM YAML FILE failed: ") + e.what());
    }
    int64_t size = fs.GetFileSize(*handle);
    if (size < 0) {
        throw BinderException(
            "FROM YAML FILE failed: GetFileSize returned -1 for '" +
            bd->file_path + "'");
    }
    // Phase 65.1 IN-04: fail-fast on hostile file sizes BEFORE allocating
    // `std::string(size, '\0')`. The Rust side enforces a 1 MiB cap via
    // `from_yaml_with_size_cap`, but without a C++ pre-check a multi-GB
    // YAML path would allocate that many bytes of std::string storage,
    // run the read, hand the buffer to Rust, and only then get rejected
    // — practical outcome on a memory-constrained host is OOM kill. The
    // C++ cap is intentionally LOOSER than the Rust cap (16 MiB vs
    // 1 MiB) so the Rust limit remains the single source of truth for
    // legitimate sizing decisions; this cap exists purely as
    // defence-in-depth against a hostile or accidentally-pathological
    // file size.
    constexpr int64_t MAX_YAML_BYTES = 16 * 1024 * 1024;
    if (size > MAX_YAML_BYTES) {
        throw BinderException(
            "FROM YAML FILE failed: file '" + bd->file_path + "' is " +
            std::to_string(size) + " bytes (max " +
            std::to_string(MAX_YAML_BYTES) + ")");
    }
    std::string yaml_content(static_cast<size_t>(size), '\0');
    int64_t got = (size == 0) ? 0
                              : fs.Read(*handle, yaml_content.data(), size);
    if (got != size) {
        throw BinderException(
            "FROM YAML FILE failed: short read (" + std::to_string(got) +
            " of " + std::to_string(size) + ") for '" + bd->file_path + "'");
    }

    // Bridge into Rust to parse + enrich + serialize. The Rust side enforces
    // YAML_SIZE_CAP (1 MiB) inside from_yaml_with_size_cap, which remains the
    // single source of truth for the legitimate size limit. The 16 MiB
    // pre-check above is defence-in-depth against a hostile file size that
    // would balloon the std::string allocation before Rust gets a chance to
    // reject it.
    char *out_ptr = nullptr;
    size_t out_len = 0;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));

    uint8_t rc = sv_compute_create_from_yaml_rust(
        reinterpret_cast<const uint8_t *>(yaml_content.data()),
        yaml_content.size(),
        reinterpret_cast<const uint8_t *>(bd->view_name.data()),
        bd->view_name.size(),
        reinterpret_cast<const uint8_t *>(bd->comment.data()),
        bd->comment.size(),
        &out_ptr,
        &out_len,
        error_buf,
        sizeof(error_buf));

    if (rc != 0 || out_ptr == nullptr) {
        // Match "FROM YAML FILE failed:" wording for size/parse errors so
        // the legacy phase53 sqllogictest assertions stay green.
        throw BinderException(
            std::string("FROM YAML FILE failed: ") + error_buf);
    }

    // Move the heap-owned UTF-8 buffer into bd->new_def, then release the
    // Rust allocation via sv_free_buffer (the exact (ptr, len) pair Rust
    // returned). The string copy is unavoidable because std::string demands
    // its own allocator-owned storage; for a 1 MiB cap this is bounded.
    bd->new_def.assign(out_ptr, out_len);
    sv_free_buffer(out_ptr, out_len);

    return_types.push_back(LogicalType::VARCHAR);
    names.push_back("new_def");
    return std::move(bd);
}

static unique_ptr<LocalTableFunctionState> sv_create_from_yaml_init_local(
    ExecutionContext & /*context*/,
    TableFunctionInitInput & /*input*/,
    GlobalTableFunctionState * /*global_state*/) {
    return make_uniq<CreateFromYamlLocalState>();
}

// Phase 65.1 D-07 (IN-01 refresh): this TF's single-shot emission is enforced
// via `init_local` — `sv_create_from_yaml_init_local` (defined just above)
// constructs the `CreateFromYamlLocalState` flag that the exec callback below
// reads to decide whether to emit. `sv_register_table_function` refuses null
// `init_cb` at registration time per Phase 65.1 D-05 (registration helper
// surfaces the refusal text "init_cb is mandatory" via the error_buf), so
// reaching the `InternalException` branch in the exec callback is a
// corrupted-state diagnostic rather than a normal failure mode. The earlier
// recommendation that previously lived here (to rewire registration through
// the init_local callback) is retired — D-05 enforcement (Plan 02) and D-04
// fallback removal (Plan 08) closed it.
static void sv_create_from_yaml_function(
    ClientContext & /*context*/,
    TableFunctionInput &data_p,
    DataChunk &output) {
    auto &bd = data_p.bind_data->Cast<CreateFromYamlBindData>();
    auto *state_p = data_p.local_state.get();
    if (state_p == nullptr) {
        // Registration refuses null init_cb at registration time per
        // Phase 65.1 D-05 (see `sv_register_table_function` in this file).
        // Reaching this branch indicates a corrupted TableFunctionInput;
        // fail loud rather than risk an unbounded row stream (Phase 65.1
        // D-04, CR-02).
        throw InternalException(
            "sv_create_from_yaml_function: local_state missing despite init_local registration");
    }
    auto &state = state_p->Cast<CreateFromYamlLocalState>();
    if (state.emitted) {
        output.SetCardinality(0);
        return;
    }
    output.SetValue(0, 0, Value(bd.new_def));
    output.SetCardinality(1);
    state.emitted = true;
}

// ---------------------------------------------------------------------------
// Shared read-side FFI wire-format readers — Phase 65 Plan 05
// ---------------------------------------------------------------------------
// Foundation shared by every migrated read-side table function: the generic
// VARCHAR / VARCHAR+BOOL scaffolds below, the `list_semantic_views` family,
// and the `semantic_view` bind. Introduced with the Wave-0
// `list_semantic_views` spike (Task 1) — the first read-side TF moved off
// duckdb-rs `register_table_function_with_extra_info` (which marshals
// `ClientContext &` away — Plan 01 Spike A6) onto the C++ Catalog API path
// via `sv_register_table_function` (Plan 04). See `65-05-SPIKE-SUMMARY.md` for
// the bridge design; the per-call `Connection probe` borrow contract is
// documented on the FFI dispatcher declarations above and applied by the
// shared bind helpers (`sv_run_varchar_bind` et al.) below.
//
// Wire format from the Rust dispatcher (AR-3 self-describing): a flat binary
// buffer beginning with a schema header of column type tags, then:
//   u32 col_count (little-endian), then col_count u8 type tags
//   u32 row_count (little-endian)
//   for each row, for each column:
//     u32 byte_len (little-endian) + byte_len bytes (UTF-8, NOT NUL-terminated)
// (a trailing BOOLEAN column, where present, is a single u8). The buffer is
// heap-owned by Rust; C++ parses it into BindData and immediately releases it
// via `sv_free_buffer`.

// Helper: read a little-endian u32 from buf[offset..offset+4] and advance.
// Throws BinderException on out-of-bounds — defensive against an FFI buffer
// truncated by a panic or allocation failure on the Rust side.
static uint32_t sv_read_u32_le(const char *buf, size_t buf_len, size_t &offset,
                               const char *fn_name) {
    if (offset + 4 > buf_len) {
        throw BinderException(
            std::string(fn_name) + ": FFI buffer truncated (expected u32 at offset " +
            std::to_string(offset) + " of " + std::to_string(buf_len) + ")");
    }
    auto p = reinterpret_cast<const unsigned char *>(buf + offset);
    uint32_t v = static_cast<uint32_t>(p[0])
               | (static_cast<uint32_t>(p[1]) << 8)
               | (static_cast<uint32_t>(p[2]) << 16)
               | (static_cast<uint32_t>(p[3]) << 24);
    offset += 4;
    return v;
}

static std::string sv_read_string(const char *buf, size_t buf_len, size_t &offset,
                                  const char *fn_name) {
    uint32_t len = sv_read_u32_le(buf, buf_len, offset, fn_name);
    if (offset + len > buf_len) {
        throw BinderException(
            std::string(fn_name) + ": FFI buffer truncated (expected " +
            std::to_string(len) + " string bytes at offset " +
            std::to_string(offset) + " of " + std::to_string(buf_len) + ")");
    }
    std::string s(buf + offset, len);
    offset += len;
    return s;
}

// Self-describing wire-format schema tags (AR-3). Kept in sync with
// `WIRE_TAG_VARCHAR` / `WIRE_TAG_BOOL` in `src/ddl/read_ffi.rs` — the Rust
// serializers emit these, the parsers below assert them.
static constexpr uint8_t SV_WIRE_VARCHAR = 1;
static constexpr uint8_t SV_WIRE_BOOL = 2;

// Read + validate the self-describing schema header (AR-3): `u32 col_count`
// followed by one `u8` type tag per column. `expected_tags` is the column
// layout the bind callback declared; a mismatch means the Rust dispatcher and
// this C++ bind disagree on the result shape — fail loudly at the header
// rather than misparse cell bytes downstream.
//
// An empty result set is serialized with `col_count == 0` and no tags (there
// are no rows, hence no cells to misalign); the schema check is skipped in
// that case. Returns the `col_count` read from the header so the caller can
// enforce the empty-encoding invariant (`col_count == 0` ⟺ `row_count == 0`)
// against the `row_count` that follows.
static uint32_t sv_read_wire_schema(const char *buf, size_t buf_len, size_t &offset,
                                    const std::vector<uint8_t> &expected_tags,
                                    const char *fn_name) {
    uint32_t col_count = sv_read_u32_le(buf, buf_len, offset, fn_name);
    if (col_count == 0) {
        return 0;  // empty result set — no schema to validate
    }
    if (col_count != expected_tags.size()) {
        throw BinderException(
            std::string(fn_name) + ": FFI wire-format column count mismatch (payload declares " +
            std::to_string(col_count) + " columns, bind expects " +
            std::to_string(expected_tags.size()) + ")");
    }
    if (offset + col_count > buf_len) {
        throw BinderException(
            std::string(fn_name) + ": FFI buffer truncated (expected " +
            std::to_string(col_count) + " schema tags at offset " +
            std::to_string(offset) + " of " + std::to_string(buf_len) + ")");
    }
    for (uint32_t c = 0; c < col_count; ++c) {
        uint8_t tag = static_cast<uint8_t>(buf[offset + c]);
        if (tag != expected_tags[c]) {
            throw BinderException(
                std::string(fn_name) + ": FFI wire-format column-type mismatch at column " +
                std::to_string(c) + " (payload tag " + std::to_string(tag) +
                ", bind expects " + std::to_string(expected_tags[c]) + ")");
        }
    }
    offset += col_count;
    return col_count;
}

// The `list_semantic_views` / `list_terse_semantic_views` table functions are
// defined below as thin adapters over the generic VARCHAR-rows scaffold. C-1
// (code-review 2026-07-11) folded the Wave-0 spike's bespoke structs,
// hand-rolled 6-string parse loop, and duplicate chunked emitter onto
// `sv_run_varchar_bind` + `sv_parse_varchar_payload` + `sv_emit_varchar_rows`
// — which also picks up the trailing-bytes wire check the bespoke parser
// lacked.

// ---------------------------------------------------------------------------
// Phase 65 Plan 05 (Tasks 2-5) — Generic VARCHAR-rows + VARCHAR/BOOL-rows
// helpers for the remaining read-side TF migrations.
// ---------------------------------------------------------------------------
//
// Most of the Phase 65 Plan 05 migrations emit homogeneous VARCHAR result
// columns (the various SHOW SEMANTIC ... / DESCRIBE SEMANTIC VIEW shapes).
// To keep the per-TF migration delta to a thin adapter, we factor the
// payload-parse + bind-data shape + exec emitter into generic helpers here.
//
// Each migrated TF still owns its own bind callback (to declare the right
// column names and call the matching `sv_<name>_bind_rust` dispatcher), but
// the bulk of the bind + the entire exec/init are reused.

struct SvVarcharBindData : public TableFunctionData {
    std::vector<std::vector<std::string>> rows;
    size_t expected_cols = 0;
};

struct SvVarcharLocalState : public LocalTableFunctionState {
    // Next row to emit. Bind-materialized row sets can exceed one DataChunk's
    // capacity (STANDARD_VECTOR_SIZE = 2048), so the exec callback emits in
    // chunks and resumes from this cursor — a single-shot `emitted` flag
    // overflowed the chunk for >2048 rows (writes past the vector's data
    // buffer; no bounds check in release builds). Shared by
    // sv_emit_varchar_rows and sv_emit_varchar_bool_rows.
    idx_t next_row = 0;
};

// Parse a length-prefixed VARCHAR wire-format payload into bd.rows. The
// expected column count is taken from bd.expected_cols (must be set by the
// caller before invocation). Throws BinderException on truncated payload.
static void sv_parse_varchar_payload(const char *buf, size_t buf_len,
                                     SvVarcharBindData &bd,
                                     const char *fn_name) {
    if (buf == nullptr) {
        return;  // rc=0 with null buffer == zero rows
    }
    size_t offset = 0;
    // Validate the self-describing schema header (AR-3): all VARCHAR columns.
    uint32_t schema_cols =
        sv_read_wire_schema(buf, buf_len, offset,
                            std::vector<uint8_t>(bd.expected_cols, SV_WIRE_VARCHAR),
                            fn_name);
    uint32_t row_count = sv_read_u32_le(buf, buf_len, offset, fn_name);
    if (schema_cols == 0 && row_count != 0) {
        throw BinderException(
            std::string(fn_name) +
            ": FFI wire-format empty-schema header (col_count == 0) with row_count == " +
            std::to_string(row_count) + " > 0");
    }
    bd.rows.reserve(row_count);
    for (uint32_t r = 0; r < row_count; ++r) {
        std::vector<std::string> row;
        row.reserve(bd.expected_cols);
        for (size_t c = 0; c < bd.expected_cols; ++c) {
            row.push_back(sv_read_string(buf, buf_len, offset, fn_name));
        }
        bd.rows.push_back(std::move(row));
    }
    if (offset != buf_len) {
        throw BinderException(
            std::string(fn_name) +
            ": FFI buffer has trailing bytes (consumed " +
            std::to_string(offset) + " of " + std::to_string(buf_len) + ")");
    }
}

static unique_ptr<LocalTableFunctionState> sv_varchar_init_local(
    ExecutionContext & /*context*/,
    TableFunctionInitInput & /*input*/,
    GlobalTableFunctionState * /*global_state*/) {
    return make_uniq<SvVarcharLocalState>();
}

static void sv_emit_varchar_rows(
    ClientContext & /*context*/,
    TableFunctionInput &data_p,
    DataChunk &output) {
    auto &bd = data_p.bind_data->Cast<SvVarcharBindData>();
    auto *state_p = data_p.local_state.get();
    if (state_p == nullptr) {
        // Registration refuses null init_cb at registration time per
        // Phase 65.1 D-05 (see `sv_register_table_function` in this file).
        // Reaching this branch indicates a corrupted TableFunctionInput;
        // fail loud rather than risk an unbounded row stream (Phase 65.1
        // D-04, CR-02).
        throw InternalException(
            "sv_emit_varchar_rows: local_state missing despite init_local registration");
    }
    auto &state = state_p->Cast<SvVarcharLocalState>();
    idx_t total = bd.rows.size();
    if (state.next_row >= total) {
        output.SetCardinality(0);
        return;
    }
    idx_t remaining = total - state.next_row;
    idx_t count = remaining < STANDARD_VECTOR_SIZE ? remaining : STANDARD_VECTOR_SIZE;
    for (idx_t i = 0; i < count; ++i) {
        const auto &row = bd.rows[state.next_row + i];
        for (size_t c = 0; c < row.size(); ++c) {
            output.SetValue(c, i, Value(row[c]));
        }
    }
    output.SetCardinality(count);
    state.next_row += count;
}

// VARCHAR-rows-with-trailing-BOOL shape (used by
// show_semantic_dimensions_for_metric, which returns 3 VARCHAR + 1 BOOLEAN).
struct SvVarcharBoolBindData : public TableFunctionData {
    std::vector<std::pair<std::vector<std::string>, bool>> rows;
    size_t expected_varchar_cols = 0;  // number of VARCHAR cells per row
};

static void sv_parse_varchar_bool_payload(const char *buf, size_t buf_len,
                                          SvVarcharBoolBindData &bd,
                                          const char *fn_name) {
    if (buf == nullptr) {
        return;
    }
    size_t offset = 0;
    // Validate the self-describing schema header (AR-3): N VARCHAR columns
    // followed by a single trailing BOOLEAN column.
    std::vector<uint8_t> expected_tags(bd.expected_varchar_cols, SV_WIRE_VARCHAR);
    expected_tags.push_back(SV_WIRE_BOOL);
    uint32_t schema_cols = sv_read_wire_schema(buf, buf_len, offset, expected_tags, fn_name);
    uint32_t row_count = sv_read_u32_le(buf, buf_len, offset, fn_name);
    if (schema_cols == 0 && row_count != 0) {
        throw BinderException(
            std::string(fn_name) +
            ": FFI wire-format empty-schema header (col_count == 0) with row_count == " +
            std::to_string(row_count) + " > 0");
    }
    bd.rows.reserve(row_count);
    for (uint32_t r = 0; r < row_count; ++r) {
        std::vector<std::string> strs;
        strs.reserve(bd.expected_varchar_cols);
        for (size_t c = 0; c < bd.expected_varchar_cols; ++c) {
            strs.push_back(sv_read_string(buf, buf_len, offset, fn_name));
        }
        if (offset + 1 > buf_len) {
            throw BinderException(
                std::string(fn_name) +
                ": FFI buffer truncated (expected u8 bool at offset " +
                std::to_string(offset) + " of " + std::to_string(buf_len) + ")");
        }
        bool b = buf[offset] != 0;
        offset += 1;
        bd.rows.emplace_back(std::move(strs), b);
    }
    if (offset != buf_len) {
        throw BinderException(
            std::string(fn_name) +
            ": FFI buffer has trailing bytes (consumed " +
            std::to_string(offset) + " of " + std::to_string(buf_len) + ")");
    }
}

static void sv_emit_varchar_bool_rows(
    ClientContext & /*context*/,
    TableFunctionInput &data_p,
    DataChunk &output) {
    auto &bd = data_p.bind_data->Cast<SvVarcharBoolBindData>();
    auto *state_p = data_p.local_state.get();
    if (state_p == nullptr) {
        // Registration refuses null init_cb at registration time per
        // Phase 65.1 D-05 (see `sv_register_table_function` in this file).
        // Reaching this branch indicates a corrupted TableFunctionInput;
        // fail loud rather than risk an unbounded row stream (Phase 65.1
        // D-04, CR-02).
        throw InternalException(
            "sv_emit_varchar_bool_rows: local_state missing despite init_local registration");
    }
    auto &state = state_p->Cast<SvVarcharLocalState>();
    idx_t total = bd.rows.size();
    if (state.next_row >= total) {
        output.SetCardinality(0);
        return;
    }
    idx_t remaining = total - state.next_row;
    idx_t count = remaining < STANDARD_VECTOR_SIZE ? remaining : STANDARD_VECTOR_SIZE;
    for (idx_t i = 0; i < count; ++i) {
        const auto &strs = bd.rows[state.next_row + i].first;
        for (size_t c = 0; c < strs.size(); ++c) {
            output.SetValue(c, i, Value(strs[c]));
        }
        // BOOLEAN trailing column at index strs.size().
        output.SetValue(strs.size(), i, Value::BOOLEAN(bd.rows[state.next_row + i].second));
    }
    output.SetCardinality(count);
    state.next_row += count;
}

// Common idiom for the bind callback: open Connection, run Rust dispatcher
// for the zero-arg case, parse payload into bd. Used by all 5 zero-arg
// migrated TFs (Task 2 / Wave 1) and reused by Task 3 / Wave 2 with extra
// VARCHAR args supplied to the dispatcher.
//
// `dispatcher` returns rc==0 on success (payload populated), rc!=0 with
// error_buf NUL-terminated otherwise. Throws BinderException on rc!=0 with
// the "<fn_name> failed: <error_buf>" message — matches the Wave 0 spike.
template <typename DispatcherFn>
static void sv_run_varchar_bind(ClientContext &context,
                                SvVarcharBindData &bd,
                                size_t expected_cols,
                                const char *fn_name,
                                DispatcherFn &&dispatcher) {
    bd.expected_cols = expected_cols;
    Connection probe(*context.db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);
    SvOwnedBuffer payload;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));
    uint8_t rc = dispatcher(borrowed,
                            &payload.ptr, &payload.len,
                            error_buf, sizeof(error_buf));
    if (rc != 0) {
        throw BinderException(std::string(fn_name) + " failed: " + error_buf);
    }
    sv_parse_varchar_payload(payload.ptr, payload.len, bd, fn_name);
}

// Serialise the caller's schema resolution order from the `search_path` named
// parameter into the shared varchar-list wire format.
//
// Absent or NULL yields an empty buffer, which the Rust side reads as "no path
// supplied" and falls back to its pre-injection rule. That matters: a table
// function invoked directly, rather than through the parser override that
// injects the path, must keep working.
// Defined further down, next to the other named-parameter marshalling it
// shares its wire format with; forward-declared here because the single-name
// bind helper below is the first caller.
static std::vector<uint8_t> sv_serialise_string_list(const Value &list_val,
                                                     const char *param_name);

static std::vector<uint8_t> sv_search_path_payload(TableFunctionBindInput &input) {
    auto it = input.named_parameters.find("search_path");
    if (it == input.named_parameters.end() || it->second.IsNull()) {
        return {};
    }
    return sv_serialise_string_list(it->second, "search_path");
}

// Variant for TFs with one VARCHAR argument (the view name). Extracts the
// name from input.inputs[0] and forwards it as (ptr, len) to the dispatcher
// alongside the borrowed connection handle.
template <typename DispatcherFn>
static void sv_run_varchar_bind_with_name(ClientContext &context,
                                          TableFunctionBindInput &input,
                                          SvVarcharBindData &bd,
                                          size_t expected_cols,
                                          const char *fn_name,
                                          DispatcherFn &&dispatcher) {
    bd.expected_cols = expected_cols;
    // FF-4: guard the positional view-name argument. Without this a NULL
    // argument (e.g. describe_semantic_view(NULL)) reached GetValue<string>()
    // and surfaced as the confusing "view 'NULL' does not exist"; match the
    // semantic_view() binder's up-front BinderException instead.
    if (input.inputs.empty() || input.inputs[0].IsNull()) {
        throw BinderException(std::string(fn_name) +
                              ": view name is required (positional arg 0)");
    }
    std::string name = input.inputs[0].GetValue<std::string>();
    std::vector<uint8_t> sp = sv_search_path_payload(input);
    Connection probe(*context.db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);
    SvOwnedBuffer payload;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));
    uint8_t rc = dispatcher(borrowed,
                            reinterpret_cast<const uint8_t *>(name.data()), name.size(),
                            sp.empty() ? nullptr : sp.data(), sp.size(),
                            &payload.ptr, &payload.len,
                            error_buf, sizeof(error_buf));
    if (rc != 0) {
        throw BinderException(std::string(fn_name) + " failed: " + error_buf);
    }
    sv_parse_varchar_payload(payload.ptr, payload.len, bd, fn_name);
}

// Variant for TFs with two VARCHAR arguments (view_name, metric_name).
template <typename DispatcherFn>
static void sv_run_varchar_bool_bind_with_two_names(
    ClientContext &context,
    TableFunctionBindInput &input,
    SvVarcharBoolBindData &bd,
    size_t expected_varchar_cols,
    const char *fn_name,
    DispatcherFn &&dispatcher) {
    bd.expected_varchar_cols = expected_varchar_cols;
    // FF-4: guard both positional arguments (view_name, metric_name) before
    // GetValue<string>() so a NULL argument raises a clear BinderException
    // rather than resolving to a spurious "'NULL' does not exist".
    if (input.inputs.size() < 2 || input.inputs[0].IsNull() ||
        input.inputs[1].IsNull()) {
        throw BinderException(
            std::string(fn_name) +
            ": view name and metric name are required (positional args 0, 1)");
    }
    std::string name1 = input.inputs[0].GetValue<std::string>();
    std::string name2 = input.inputs[1].GetValue<std::string>();
    std::vector<uint8_t> sp = sv_search_path_payload(input);
    Connection probe(*context.db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);
    SvOwnedBuffer payload;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));
    uint8_t rc = dispatcher(borrowed,
                            reinterpret_cast<const uint8_t *>(name1.data()), name1.size(),
                            reinterpret_cast<const uint8_t *>(name2.data()), name2.size(),
                            sp.empty() ? nullptr : sp.data(), sp.size(),
                            &payload.ptr, &payload.len,
                            error_buf, sizeof(error_buf));
    if (rc != 0) {
        throw BinderException(std::string(fn_name) + " failed: " + error_buf);
    }
    sv_parse_varchar_bool_payload(payload.ptr, payload.len, bd, fn_name);
}

template <typename DispatcherFn>
static void sv_run_varchar_bool_bind(ClientContext &context,
                                     SvVarcharBoolBindData &bd,
                                     size_t expected_varchar_cols,
                                     const char *fn_name,
                                     DispatcherFn &&dispatcher) {
    bd.expected_varchar_cols = expected_varchar_cols;
    Connection probe(*context.db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);
    SvOwnedBuffer payload;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));
    uint8_t rc = dispatcher(borrowed,
                            &payload.ptr, &payload.len,
                            error_buf, sizeof(error_buf));
    if (rc != 0) {
        throw BinderException(std::string(fn_name) + " failed: " + error_buf);
    }
    sv_parse_varchar_bool_payload(payload.ptr, payload.len, bd, fn_name);
}

// ---------------------------------------------------------------------------
// list_semantic_views — Phase 65 Plan 05 Task 1 (Wave 0 spike), folded onto
// the generic scaffold (C-1, code-review 2026-07-11)
// ---------------------------------------------------------------------------
// 6-column VARCHAR: created_on, name, kind, database_name, schema_name,
// comment. Superset of list_terse_semantic_views (which drops `comment`);
// both share the Rust body `list_view_rows` and the same wire format, so the
// strict generic parser (`sv_parse_varchar_payload`, incl. the trailing-bytes
// check) applies uniformly.

static unique_ptr<FunctionData> sv_list_semantic_views_bind(
    ClientContext &context,
    TableFunctionBindInput & /*input*/,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    // Column layout must match the v0.9.0 Rust VTab exactly — SELECT * FROM
    // list_semantic_views() across the suite relies on byte-identical names
    // and order.
    static const char *const COL_NAMES[] = {
        "created_on", "name", "kind", "database_name", "schema_name", "comment",
    };
    for (auto cn : COL_NAMES) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind(
        context, *bd, /*expected_cols*/ 6, "list_semantic_views",
        [](duckdb_connection borrowed, char **out_ptr, size_t *out_len,
           char *error_buf, size_t error_buf_len) {
            return sv_list_semantic_views_bind_rust(
                borrowed, out_ptr, out_len, error_buf, error_buf_len);
        });
    return std::move(bd);
}

extern "C" {
    bool sv_register_list_semantic_views(duckdb_database db_handle,
                                         char *error_buf, size_t error_buf_len) {
        // Zero-argument table function — no arg_types array.
        return sv_register_table_function(
            db_handle,
            "list_semantic_views",
            /*arg_types*/ nullptr, /*arg_count*/ 0,
            sv_list_semantic_views_bind,
            sv_emit_varchar_rows,
            sv_varchar_init_local,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// list_terse_semantic_views — Phase 65 Plan 05 Task 2 (Wave 1)
// ---------------------------------------------------------------------------
// 5-column subset of list_semantic_views — same bridge mechanism, no
// per-function BindData struct (uses generic SvVarcharBindData via the
// sv_run_varchar_bind helper).
//
// Columns: created_on, name, kind, database_name, schema_name.

static unique_ptr<FunctionData> sv_list_terse_semantic_views_bind(
    ClientContext &context,
    TableFunctionBindInput & /*input*/,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COL_NAMES[] = {
        "created_on", "name", "kind", "database_name", "schema_name",
    };
    for (auto cn : COL_NAMES) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind(
        context, *bd, /*expected_cols*/ 5, "list_terse_semantic_views",
        [](duckdb_connection borrowed, char **out_ptr, size_t *out_len,
           char *error_buf, size_t error_buf_len) {
            return sv_list_terse_semantic_views_bind_rust(
                borrowed, out_ptr, out_len, error_buf, error_buf_len);
        });
    return std::move(bd);
}

extern "C" {
    bool sv_register_list_terse_semantic_views(duckdb_database db_handle,
                                               char *error_buf, size_t error_buf_len) {
        return sv_register_table_function(
            db_handle,
            "list_terse_semantic_views",
            /*arg_types*/ nullptr, /*arg_count*/ 0,
            sv_list_terse_semantic_views_bind,
            sv_emit_varchar_rows,
            sv_varchar_init_local,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// show_semantic_dimensions_all — Phase 65 Plan 05 Task 2 (Wave 1)
// ---------------------------------------------------------------------------
// 8-column VARCHAR: database_name, schema_name, semantic_view_name,
// table_name, name, data_type, synonyms, comment.

static unique_ptr<FunctionData> sv_show_semantic_dimensions_all_bind(
    ClientContext &context,
    TableFunctionBindInput & /*input*/,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name", "table_name",
        "name", "data_type", "synonyms", "comment",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind(
        context, *bd, /*expected_cols*/ 8, "show_semantic_dimensions_all",
        [](duckdb_connection borrowed, char **out_ptr, size_t *out_len,
           char *error_buf, size_t error_buf_len) {
            return sv_show_semantic_dimensions_all_bind_rust(
                borrowed, out_ptr, out_len, error_buf, error_buf_len);
        });
    return std::move(bd);
}

extern "C" {
    bool sv_register_show_semantic_dimensions_all(duckdb_database db_handle,
                                                  char *error_buf, size_t error_buf_len) {
        return sv_register_table_function(
            db_handle, "show_semantic_dimensions_all",
            nullptr, 0,
            sv_show_semantic_dimensions_all_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// show_semantic_metrics_all — Phase 65 Plan 05 Task 2 (Wave 1)
// ---------------------------------------------------------------------------
// 8-column VARCHAR (same schema as show_semantic_dimensions_all).

static unique_ptr<FunctionData> sv_show_semantic_metrics_all_bind(
    ClientContext &context,
    TableFunctionBindInput & /*input*/,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name", "table_name",
        "name", "data_type", "synonyms", "comment",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind(
        context, *bd, 8, "show_semantic_metrics_all",
        [](duckdb_connection borrowed, char **out_ptr, size_t *out_len,
           char *error_buf, size_t error_buf_len) {
            return sv_show_semantic_metrics_all_bind_rust(
                borrowed, out_ptr, out_len, error_buf, error_buf_len);
        });
    return std::move(bd);
}

extern "C" {
    bool sv_register_show_semantic_metrics_all(duckdb_database db_handle,
                                               char *error_buf, size_t error_buf_len) {
        return sv_register_table_function(
            db_handle, "show_semantic_metrics_all",
            nullptr, 0,
            sv_show_semantic_metrics_all_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// show_semantic_facts_all — Phase 65 Plan 05 Task 2 (Wave 1)
// ---------------------------------------------------------------------------
// 8-column VARCHAR (same schema as show_semantic_dimensions_all).

static unique_ptr<FunctionData> sv_show_semantic_facts_all_bind(
    ClientContext &context,
    TableFunctionBindInput & /*input*/,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name", "table_name",
        "name", "data_type", "synonyms", "comment",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind(
        context, *bd, 8, "show_semantic_facts_all",
        [](duckdb_connection borrowed, char **out_ptr, size_t *out_len,
           char *error_buf, size_t error_buf_len) {
            return sv_show_semantic_facts_all_bind_rust(
                borrowed, out_ptr, out_len, error_buf, error_buf_len);
        });
    return std::move(bd);
}

extern "C" {
    bool sv_register_show_semantic_facts_all(duckdb_database db_handle,
                                             char *error_buf, size_t error_buf_len) {
        return sv_register_table_function(
            db_handle, "show_semantic_facts_all",
            nullptr, 0,
            sv_show_semantic_facts_all_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// show_semantic_materializations_all — Phase 65 Plan 05 Task 2 (Wave 1)
// ---------------------------------------------------------------------------
// 7-column VARCHAR: database_name, schema_name, semantic_view_name, name,
// table, dimensions, metrics.

static unique_ptr<FunctionData> sv_show_semantic_materializations_all_bind(
    ClientContext &context,
    TableFunctionBindInput & /*input*/,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name",
        "name", "table", "dimensions", "metrics",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind(
        context, *bd, 7, "show_semantic_materializations_all",
        [](duckdb_connection borrowed, char **out_ptr, size_t *out_len,
           char *error_buf, size_t error_buf_len) {
            return sv_show_semantic_materializations_all_bind_rust(
                borrowed, out_ptr, out_len, error_buf, error_buf_len);
        });
    return std::move(bd);
}

extern "C" {
    bool sv_register_show_semantic_materializations_all(duckdb_database db_handle,
                                                        char *error_buf, size_t error_buf_len) {
        return sv_register_table_function(
            db_handle, "show_semantic_materializations_all",
            nullptr, 0,
            sv_show_semantic_materializations_all_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// Wave 2 single-arg TFs — Phase 65 Plan 05 Task 3
// ---------------------------------------------------------------------------
// show_columns_in_semantic_view, describe_semantic_view, and the 4
// single-view show_semantic_<entity> variants all take one VARCHAR arg
// (the view name) and emit homogeneous VARCHAR rows. The single
// show_semantic_dimensions_for_metric takes two VARCHAR args and emits
// VARCHAR + trailing BOOL rows (handled in the dedicated block further
// down).

static unique_ptr<FunctionData> sv_show_columns_in_semantic_view_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name", "column_name",
        "data_type", "kind", "expression", "comment",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind_with_name(
        context, input, *bd, 8, "show_columns_in_semantic_view",
        [](duckdb_connection borrowed,
           const uint8_t *np, size_t nl,
           const uint8_t *sp, size_t spl,
           char **op, size_t *ol, char *eb, size_t ebl) {
            return sv_show_columns_in_semantic_view_bind_rust(
                borrowed, np, nl, sp, spl, op, ol, eb, ebl);
        });
    return std::move(bd);
}

static unique_ptr<FunctionData> sv_describe_semantic_view_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "object_kind", "object_name", "parent_entity", "property", "property_value",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind_with_name(
        context, input, *bd, 5, "describe_semantic_view",
        [](duckdb_connection borrowed,
           const uint8_t *np, size_t nl,
           const uint8_t *sp, size_t spl,
           char **op, size_t *ol, char *eb, size_t ebl) {
            return sv_describe_semantic_view_bind_rust(
                borrowed, np, nl, sp, spl, op, ol, eb, ebl);
        });
    return std::move(bd);
}

static unique_ptr<FunctionData> sv_show_semantic_dimensions_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name", "table_name",
        "name", "data_type", "synonyms", "comment",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind_with_name(
        context, input, *bd, 8, "show_semantic_dimensions",
        [](duckdb_connection borrowed,
           const uint8_t *np, size_t nl,
           const uint8_t *sp, size_t spl,
           char **op, size_t *ol, char *eb, size_t ebl) {
            return sv_show_semantic_dimensions_bind_rust(
                borrowed, np, nl, sp, spl, op, ol, eb, ebl);
        });
    return std::move(bd);
}

static unique_ptr<FunctionData> sv_show_semantic_metrics_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name", "table_name",
        "name", "data_type", "synonyms", "comment",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind_with_name(
        context, input, *bd, 8, "show_semantic_metrics",
        [](duckdb_connection borrowed,
           const uint8_t *np, size_t nl,
           const uint8_t *sp, size_t spl,
           char **op, size_t *ol, char *eb, size_t ebl) {
            return sv_show_semantic_metrics_bind_rust(
                borrowed, np, nl, sp, spl, op, ol, eb, ebl);
        });
    return std::move(bd);
}

static unique_ptr<FunctionData> sv_show_semantic_facts_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name", "table_name",
        "name", "data_type", "synonyms", "comment",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind_with_name(
        context, input, *bd, 8, "show_semantic_facts",
        [](duckdb_connection borrowed,
           const uint8_t *np, size_t nl,
           const uint8_t *sp, size_t spl,
           char **op, size_t *ol, char *eb, size_t ebl) {
            return sv_show_semantic_facts_bind_rust(
                borrowed, np, nl, sp, spl, op, ol, eb, ebl);
        });
    return std::move(bd);
}

static unique_ptr<FunctionData> sv_show_semantic_materializations_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    static const char *const COLS[] = {
        "database_name", "schema_name", "semantic_view_name",
        "name", "table", "dimensions", "metrics",
    };
    for (auto cn : COLS) {
        return_types.push_back(LogicalType::VARCHAR);
        names.emplace_back(cn);
    }
    sv_run_varchar_bind_with_name(
        context, input, *bd, 7, "show_semantic_materializations",
        [](duckdb_connection borrowed,
           const uint8_t *np, size_t nl,
           const uint8_t *sp, size_t spl,
           char **op, size_t *ol, char *eb, size_t ebl) {
            return sv_show_semantic_materializations_bind_rust(
                borrowed, np, nl, sp, spl, op, ol, eb, ebl);
        });
    return std::move(bd);
}

// show_semantic_dimensions_for_metric: 3 VARCHAR + 1 BOOL output, 2 VARCHAR
// input args.
static unique_ptr<FunctionData> sv_show_semantic_dimensions_for_metric_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBoolBindData>();
    return_types.push_back(LogicalType::VARCHAR);
    names.emplace_back("table_name");
    return_types.push_back(LogicalType::VARCHAR);
    names.emplace_back("name");
    return_types.push_back(LogicalType::VARCHAR);
    names.emplace_back("data_type");
    return_types.push_back(LogicalType::BOOLEAN);
    names.emplace_back("required");
    sv_run_varchar_bool_bind_with_two_names(
        context, input, *bd, /*expected_varchar_cols*/ 3,
        "show_semantic_dimensions_for_metric",
        [](duckdb_connection borrowed,
           const uint8_t *vn, size_t vnl,
           const uint8_t *mn, size_t mnl,
           const uint8_t *sp, size_t spl,
           char **op, size_t *ol, char *eb, size_t ebl) {
            return sv_show_semantic_dimensions_for_metric_bind_rust(
                borrowed, vn, vnl, mn, mnl, sp, spl, op, ol, eb, ebl);
        });
    return std::move(bd);
}

extern "C" {
    bool sv_register_show_columns_in_semantic_view(duckdb_database db_handle,
                                                   char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR};
        return sv_register_table_function(
            db_handle, "show_columns_in_semantic_view",
            args, 1,
            sv_show_columns_in_semantic_view_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
    bool sv_register_describe_semantic_view(duckdb_database db_handle,
                                            char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR};
        return sv_register_table_function(
            db_handle, "describe_semantic_view",
            args, 1,
            sv_describe_semantic_view_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
    bool sv_register_show_semantic_dimensions(duckdb_database db_handle,
                                              char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR};
        return sv_register_table_function(
            db_handle, "show_semantic_dimensions",
            args, 1,
            sv_show_semantic_dimensions_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
    bool sv_register_show_semantic_metrics(duckdb_database db_handle,
                                           char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR};
        return sv_register_table_function(
            db_handle, "show_semantic_metrics",
            args, 1,
            sv_show_semantic_metrics_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
    bool sv_register_show_semantic_facts(duckdb_database db_handle,
                                         char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR};
        return sv_register_table_function(
            db_handle, "show_semantic_facts",
            args, 1,
            sv_show_semantic_facts_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
    bool sv_register_show_semantic_materializations(duckdb_database db_handle,
                                                    char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR};
        return sv_register_table_function(
            db_handle, "show_semantic_materializations",
            args, 1,
            sv_show_semantic_materializations_bind,
            sv_emit_varchar_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
    bool sv_register_show_semantic_dimensions_for_metric(duckdb_database db_handle,
                                                         char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR, LogicalType::VARCHAR};
        return sv_register_table_function(
            db_handle, "show_semantic_dimensions_for_metric",
            args, 2,
            sv_show_semantic_dimensions_for_metric_bind,
            sv_emit_varchar_bool_rows, sv_varchar_init_local,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// Phase 65 Plan 05 Task 4 (Wave 3) — scalar function migrations
// ---------------------------------------------------------------------------
//
// Migrate the two read-side scalars (`get_ddl`, `read_yaml_from_semantic_view`)
// from duckdb-rs `register_scalar_function_with_state` to the C++ Catalog
// API path established by `sv_register_scalar_function` (Wave 1 prep). Each
// exec callback opens a per-call `Connection probe(*state.GetContext().db)`
// (the scalar analog of the bind-side `Connection(*context.db)` used by the
// 15 TF migrations) and bridges to a Rust dispatcher that performs the
// lookup + render on the borrowed connection. The borrow contract (Rust
// MUST NOT call `duckdb_disconnect`) is identical to the TF dispatchers —
// see `src/ddl/read_ffi.rs` module docs.
//
// Per-chunk Connection construction (one `Connection probe` per exec call,
// reused across every row in the chunk) keeps the dispatcher shape uniform
// with the TF migrations: each call into the C++ exec callback gets a
// fresh stack-owned Connection bound to the caller's Database, and every
// row in the chunk borrows that same handle through the reinterpret_cast
// bridge. Scalar usage in practice is one or two rows
// (`SELECT GET_DDL(...)`), so amortising the Connection ctor across the
// chunk is essentially free even at chunk size 1 — and the ctor itself is
// sub-millisecond per the READ-PATH-SPIKE evidence.

// Common helper: copy a Rust-owned UTF-8 buffer into the result Vector at
// `row_idx` via StringVector::AddString (which allocates inside the Vector's
// string heap). Throws InvalidInputException on rc!=0 with the dispatcher's
// error_buf as the message. The owned buffer is always freed via
// `sv_free_buffer` regardless of rc.
template <typename DispatcherFn>
static void sv_emit_scalar_row(Vector &result, idx_t row_idx,
                               const char *fn_name,
                               DispatcherFn &&dispatcher) {
    SvOwnedBuffer payload;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));
    uint8_t rc = dispatcher(&payload.ptr, &payload.len,
                            error_buf, sizeof(error_buf));
    if (rc != 0) {
        throw InvalidInputException(std::string(fn_name) + ": " + error_buf);
    }
    // payload.ptr is non-null on rc==0 (publish_owned_buffer guarantees this
    // when out_ptr is non-null); len may be 0 for an empty string. The null
    // branch is purely defensive, but AddString(data, len) copies `len` bytes,
    // so the fallback length MUST be 0 too — otherwise a (nullptr, len>0)
    // payload would read `len` bytes past the 1-byte "" literal (C-9,
    // code-review 2026-07-11).
    const char *out_data = payload.ptr == nullptr ? "" : payload.ptr;
    idx_t out_len = payload.ptr == nullptr ? 0 : payload.len;
    string_t out_value = StringVector::AddString(result, out_data, out_len);
    FlatVector::GetData<string_t>(result)[row_idx] = out_value;
}

// get_ddl(object_type VARCHAR, name VARCHAR [, use_fully_qualified_names
// BOOLEAN]) -> VARCHAR
//
// Both arities land here; `args.ColumnCount()` distinguishes them. The third
// argument mirrors Snowflake's `use_fully_qualified_names`: when true the
// rendered CREATE name carries the view's own schema, so re-running the output
// restores the view where it came from rather than into the executing
// session's schema.
static void sv_get_ddl_exec(DataChunk &args, ExpressionState &state,
                            Vector &result) {
    const bool has_qualify_arg = args.ColumnCount() >= 3;
    auto &type_vec = args.data[0];
    auto &name_vec = args.data[1];
    type_vec.Flatten(args.size());
    name_vec.Flatten(args.size());
    auto type_data = FlatVector::GetData<string_t>(type_vec);
    auto name_data = FlatVector::GetData<string_t>(name_vec);
    auto &type_validity = FlatVector::Validity(type_vec);
    auto &name_validity = FlatVector::Validity(name_vec);

    // Only touched when the third arity was bound; `data[2]` does not exist on
    // the two-argument call.
    bool *qualify_data = nullptr;
    if (has_qualify_arg) {
        auto &qualify_vec = args.data[2];
        qualify_vec.Flatten(args.size());
        qualify_data = FlatVector::GetData<bool>(qualify_vec);
    }

    auto &result_validity = FlatVector::Validity(result);

    Connection probe(*state.GetContext().db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);

    for (idx_t i = 0; i < args.size(); ++i) {
        // A NULL in any argument yields a NULL row, the third included: asked
        // "qualified or not?" with NULL there is no answer, and defaulting to
        // either spelling would be a guess.
        if (!type_validity.RowIsValid(i) || !name_validity.RowIsValid(i) ||
            (has_qualify_arg && !FlatVector::Validity(args.data[2]).RowIsValid(i))) {
            result_validity.SetInvalid(i);
            continue;
        }
        const string_t &t = type_data[i];
        const string_t &n = name_data[i];
        const uint8_t qualify =
            (has_qualify_arg && qualify_data[i]) ? uint8_t(1) : uint8_t(0);
        sv_emit_scalar_row(
            result, i, "get_ddl",
            [&](char **op, size_t *ol, char *eb, size_t ebl) {
                return sv_get_ddl_exec_rust(
                    borrowed,
                    reinterpret_cast<const uint8_t *>(t.GetData()), t.GetSize(),
                    reinterpret_cast<const uint8_t *>(n.GetData()), n.GetSize(),
                    // No search path: GET_DDL / READ_YAML are SCALAR
                    // functions, so the path cannot arrive as a named
                    // parameter. Passing (nullptr, 0) selects the Rust side's
                    // no-path fallback — the unique match, or an actionable
                    // ambiguity error. Conservative rather than wrong: it never
                    // returns another schema's definition. Tracked in
                    // TECH-DEBT #25. (The overload below adds an *arity*, not a
                    // named parameter, so it does not carry a path either.)
                    nullptr, 0,
                    qualify,
                    op, ol, eb, ebl);
            });
    }
    if (args.AllConstant()) {
        result.SetVectorType(VectorType::CONSTANT_VECTOR);
    }
}

// read_yaml_from_semantic_view(name VARCHAR) -> VARCHAR
static void sv_read_yaml_from_semantic_view_exec(DataChunk &args,
                                                 ExpressionState &state,
                                                 Vector &result) {
    auto &name_vec = args.data[0];
    name_vec.Flatten(args.size());
    auto name_data = FlatVector::GetData<string_t>(name_vec);
    auto &name_validity = FlatVector::Validity(name_vec);
    auto &result_validity = FlatVector::Validity(result);

    Connection probe(*state.GetContext().db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);

    for (idx_t i = 0; i < args.size(); ++i) {
        if (!name_validity.RowIsValid(i)) {
            result_validity.SetInvalid(i);
            continue;
        }
        const string_t &n = name_data[i];
        sv_emit_scalar_row(
            result, i, "read_yaml_from_semantic_view",
            [&](char **op, size_t *ol, char *eb, size_t ebl) {
                return sv_read_yaml_from_semantic_view_exec_rust(
                    borrowed,
                    reinterpret_cast<const uint8_t *>(n.GetData()), n.GetSize(),
                    // No search path: GET_DDL / READ_YAML are SCALAR
                    // functions, so the path cannot arrive as a named
                    // parameter and would need a second registered arity
                    // (a ScalarFunctionSet overload). Passing (nullptr, 0)
                    // selects the Rust side's no-path fallback — the unique
                    // match, or an actionable ambiguity error. Conservative
                    // rather than wrong: it never returns another schema's
                    // definition. Tracked in TECH-DEBT #25.
                    nullptr, 0,
                    op, ol, eb, ebl);
            });
    }
    if (args.AllConstant()) {
        result.SetVectorType(VectorType::CONSTANT_VECTOR);
    }
}

extern "C" {
    // Two arities on one name (TECH-DEBT #25): the third `BOOLEAN` argument is
    // Snowflake's `use_fully_qualified_names`. Registering a set rather than
    // two separate functions is what lets DuckDB's binder pick by argument
    // count; both overloads share `sv_get_ddl_exec`, which reads the third
    // column only when it was bound.
    bool sv_register_get_ddl(duckdb_database db_handle,
                             char *error_buf, size_t error_buf_len) {
        try {
            ScalarFunctionSet set("get_ddl");
            set.AddFunction(ScalarFunction(
                "get_ddl", {LogicalType::VARCHAR, LogicalType::VARCHAR},
                LogicalType::VARCHAR, sv_get_ddl_exec));
            set.AddFunction(ScalarFunction(
                "get_ddl",
                {LogicalType::VARCHAR, LogicalType::VARCHAR,
                 LogicalType::BOOLEAN},
                LogicalType::VARCHAR, sv_get_ddl_exec));
            return sv_register_scalar_function_set(
                db_handle, "get_ddl", std::move(set), error_buf, error_buf_len);
        } catch (const std::exception &e) {
            if (error_buf != nullptr && error_buf_len > 0) {
                snprintf(error_buf, error_buf_len,
                    "sv_register_get_ddl failed: %s", e.what());
            }
            return false;
        } catch (...) {
            if (error_buf != nullptr && error_buf_len > 0) {
                snprintf(error_buf, error_buf_len,
                    "sv_register_get_ddl failed: unknown C++ exception");
            }
            return false;
        }
    }
    bool sv_register_read_yaml_from_semantic_view(duckdb_database db_handle,
                                                  char *error_buf, size_t error_buf_len) {
        LogicalType args[] = {LogicalType::VARCHAR};
        return sv_register_scalar_function(
            db_handle, "read_yaml_from_semantic_view",
            args, 1,
            LogicalType::VARCHAR,
            sv_read_yaml_from_semantic_view_exec,
            error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// Phase 65 Plan 05 Task 5 (Wave 5) — explain_semantic_view migration
// ---------------------------------------------------------------------------
//
// Migrates `explain_semantic_view(view_name, dimensions := [...],
// metrics := [...], facts := [...])` off duckdb-rs's
// `register_table_function_with_extra_info` (which marshals
// `ClientContext &` away) to the C++ Catalog API path. The bind callback
// opens a per-call `Connection probe(*context.db)` and bridges to the
// Rust dispatcher `sv_explain_semantic_view_bind_rust` which runs the
// catalog lookup + expand + EXPLAIN on the per-call connection. Output is
// one VARCHAR row per explain-output line — reuses the Wave 1/2
// `SvVarcharBindData` shape + `sv_emit_varchar_rows` exec.
//
// Named LIST(VARCHAR) parameter handling: the three optional named
// parameters are flattened on the C++ side using
// `sv_serialise_string_list` (length-prefixed wire format) and passed as
// (ptr, len) pairs to the Rust dispatcher. Missing named parameters are
// passed as nullptr+0, which the Rust side treats as an empty list.

// Serialise a LIST(VARCHAR) Value into the standard length-prefixed wire
// format (`u32 count; for each: u32 byte_len + bytes`). The returned bytes
// can be handed directly to the Rust dispatcher as (ptr, len). Throws
// BinderException if the Value is not a LIST or contains non-VARCHAR
// children — defensive: DuckDB's named-parameter type-check already
// enforces the LIST(VARCHAR) declaration at registration time, so a
// mismatch here is a planner bug.
static std::vector<uint8_t> sv_serialise_string_list(
    const Value &list_val, const char *param_name) {
    std::vector<uint8_t> buf;
    const auto &children = ListValue::GetChildren(list_val);
    uint32_t count = static_cast<uint32_t>(children.size());
    buf.reserve(4 + children.size() * 16);
    buf.push_back(static_cast<uint8_t>(count & 0xff));
    buf.push_back(static_cast<uint8_t>((count >> 8) & 0xff));
    buf.push_back(static_cast<uint8_t>((count >> 16) & 0xff));
    buf.push_back(static_cast<uint8_t>((count >> 24) & 0xff));
    for (const auto &c : children) {
        if (c.IsNull()) {
            throw BinderException(
                std::string("explain_semantic_view: `") + param_name +
                "` contains a NULL element (only non-NULL VARCHARs accepted)");
        }
        // c.GetValue<std::string>() applies any necessary cast. The named-
        // parameter declaration is LIST(VARCHAR) so this should be a no-op.
        std::string s = c.GetValue<std::string>();
        uint32_t len = static_cast<uint32_t>(s.size());
        buf.push_back(static_cast<uint8_t>(len & 0xff));
        buf.push_back(static_cast<uint8_t>((len >> 8) & 0xff));
        buf.push_back(static_cast<uint8_t>((len >> 16) & 0xff));
        buf.push_back(static_cast<uint8_t>((len >> 24) & 0xff));
        buf.insert(buf.end(), s.begin(), s.end());
    }
    return buf;
}

static unique_ptr<FunctionData> sv_explain_semantic_view_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    auto bd = make_uniq<SvVarcharBindData>();
    bd->expected_cols = 1;
    return_types.push_back(LogicalType::VARCHAR);
    names.emplace_back("explain_output");

    if (input.inputs.empty() || input.inputs[0].IsNull()) {
        throw BinderException(
            "explain_semantic_view: view name is required (positional arg 0)");
    }
    std::string view_name = input.inputs[0].GetValue<std::string>();

    // Pull the three optional named LIST(VARCHAR) parameters. The
    // `input.named_parameters` map is case-insensitive (per
    // case_insensitive_map_t). A missing entry means the user did not
    // supply that named parameter — pass nullptr+0 to the Rust side.
    std::vector<uint8_t> dims_buf, metrics_buf, facts_buf;
    auto it_d = input.named_parameters.find("dimensions");
    if (it_d != input.named_parameters.end() && !it_d->second.IsNull()) {
        dims_buf = sv_serialise_string_list(it_d->second, "dimensions");
    }
    auto it_m = input.named_parameters.find("metrics");
    if (it_m != input.named_parameters.end() && !it_m->second.IsNull()) {
        metrics_buf = sv_serialise_string_list(it_m->second, "metrics");
    }
    auto it_f = input.named_parameters.find("facts");
    if (it_f != input.named_parameters.end() && !it_f->second.IsNull()) {
        facts_buf = sv_serialise_string_list(it_f->second, "facts");
    }

    // Scalar VARCHAR, not a list — passed to Rust as a plain UTF-8 ptr/len.
    // Absent or NULL yields (nullptr, 0), which the Rust side reads as "no
    // predicate".
    std::string where_clause;
    auto it_w = input.named_parameters.find("where_clause");
    if (it_w != input.named_parameters.end() && !it_w->second.IsNull()) {
        where_clause = it_w->second.GetValue<std::string>();
    }
    std::vector<uint8_t> sp_buf = sv_search_path_payload(input);

    Connection probe(*context.db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);

    SvOwnedBuffer payload;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));

    uint8_t rc = sv_explain_semantic_view_bind_rust(
        borrowed,
        reinterpret_cast<const uint8_t *>(view_name.data()), view_name.size(),
        sp_buf.empty()      ? nullptr : sp_buf.data(),      sp_buf.size(),
        dims_buf.empty()    ? nullptr : dims_buf.data(),    dims_buf.size(),
        metrics_buf.empty() ? nullptr : metrics_buf.data(), metrics_buf.size(),
        facts_buf.empty()   ? nullptr : facts_buf.data(),   facts_buf.size(),
        where_clause.empty() ? nullptr : reinterpret_cast<const uint8_t *>(where_clause.data()),
        where_clause.size(),
        &payload.ptr, &payload.len,
        error_buf, sizeof(error_buf));

    if (rc != 0) {
        throw BinderException(std::string("explain_semantic_view: ") + error_buf);
    }
    sv_parse_varchar_payload(payload.ptr, payload.len, *bd,
                             "explain_semantic_view");
    return std::move(bd);
}

// explain_semantic_view takes named LIST(VARCHAR) parameters (`dimensions`,
// `metrics`, `facts`) so DuckDB's binder type-checks `dimensions := [...]`.
// It builds a spec carrying those named params (from the single shared
// `sv_semantic_named_params`) plus an init_local state constructor, and
// registers via `sv_register_table_function_core` (C-7) — the hand-rolled
// TableFunction construction it used to duplicate now lives once, in the core.
// Borrow contract is unchanged: the bind receives `ClientContext &` and opens
// per-call Connections; no long-lived extension-owned connection. Failures
// surface through the forwarded `error_buf` (D-08/D-09).
static bool sv_register_explain_semantic_view_impl(duckdb_database db_handle,
                                                   char *error_buf,
                                                   size_t error_buf_len) {
    const LogicalType arg_types[] = {LogicalType::VARCHAR};
    SvTableFunctionSpec spec;
    spec.name = "explain_semantic_view";
    spec.arg_types = arg_types;
    spec.arg_count = 1;
    spec.named_params = sv_semantic_named_params();
    spec.bind_cb = sv_explain_semantic_view_bind;
    spec.exec_cb = sv_emit_varchar_rows;
    spec.init_local_cb = sv_varchar_init_local;
    spec.init_global_cb = nullptr;
    return sv_register_table_function_core(
        db_handle, spec, "sv_register_explain_semantic_view", error_buf,
        error_buf_len);
}

extern "C" {
    bool sv_register_explain_semantic_view(duckdb_database db_handle,
                                           char *error_buf, size_t error_buf_len) {
        return sv_register_explain_semantic_view_impl(
            db_handle, error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// Phase 65 Plan 05 Task 6 (Wave 6) — semantic_view migration
// ---------------------------------------------------------------------------
//
// Migrates the main expansion-path `semantic_view(view_name, dimensions :=
// [...], metrics := [...], facts := [...])` table function off duckdb-rs's
// `register_table_function_with_extra_info` (which marshals
// `ClientContext &` away and consumed the long-lived H2 query_conn) to
// the C++ Catalog API path. The bind callback opens a per-call
// `Connection probe(*context.db)` and dispatches to
// `sv_semantic_view_bind_rust` for catalog lookup + expand + LIMIT-0 type
// inference + execution-SQL construction.
//
// Streaming model: bind returns a column schema + the execution SQL
// string in BindData. `init_global` opens its OWN per-call
// `Connection probe(*context.db)`, runs `probe.Query(execution_sql)` to
// materialise the result, and stashes the
// `unique_ptr<MaterializedQueryResult>` inside the GlobalState. The
// Connection drops at end of init_global; the materialised result is
// self-contained (`ColumnDataCollection` owns its storage) and lives
// for the rest of the query. `func()` then fetches chunks via
// `result->Fetch()` and copies vector data into the output DataChunk.
//
// Per-call lifecycle: TWO Connections are constructed per
// `semantic_view(...)` invocation — one for bind (catalog probe +
// LIMIT 0 type inference), one for init_global (actual query). Both
// drop before any exec call runs. No long-lived extension-owned
// `duckdb_connection` is consumed by this path.
//
// Borrow contract: identical to the 15 prior migrations. The Rust
// dispatcher MUST NOT call `duckdb_disconnect` on the borrowed handle —
// teardown is the C++ scope's responsibility.

struct SemanticViewColumnInfo {
    std::string name;
    uint32_t type_id = 0;  // C-API DUCKDB_TYPE_* enum value (normalised:
                           // HUGEINT→BIGINT, UHUGEINT→UBIGINT).
};

// Map a C-API `DUCKDB_TYPE` enum value (which is what
// `duckdb_column_type` / `ffi::duckdb_column_type` return in Rust) to a
// C++ `LogicalType`. The two enum spaces have DIFFERENT integer values
// (e.g. C-API DUCKDB_TYPE_DECIMAL=19 vs C++ LogicalTypeId::DECIMAL=21),
// so a naive `static_cast<LogicalTypeId>(c_type_id)` would silently
// mis-type columns. This helper is the single source of truth for the
// conversion and mirrors `type_from_duckdb_type_u32` /
// `declare_output_type` in `src/query/table_function.rs`.
//
// DECIMAL / LIST / ENUM intentionally return a default (precisionless)
// LogicalType — callers (see `sv_resolve_output_logical_types`) override
// with the probed LogicalType from the LIMIT-0 result so width/scale/
// child-type are preserved.
static LogicalType sv_logical_type_from_c_type_id(uint32_t c_type_id) {
    using duckdb::LogicalTypeId;
    switch (c_type_id) {
        case DUCKDB_TYPE_INVALID:    return LogicalType::VARCHAR;  // fallback
        case DUCKDB_TYPE_BOOLEAN:    return LogicalType::BOOLEAN;
        case DUCKDB_TYPE_TINYINT:    return LogicalType::TINYINT;
        case DUCKDB_TYPE_SMALLINT:   return LogicalType::SMALLINT;
        case DUCKDB_TYPE_INTEGER:    return LogicalType::INTEGER;
        case DUCKDB_TYPE_BIGINT:     return LogicalType::BIGINT;
        case DUCKDB_TYPE_UTINYINT:   return LogicalType::UTINYINT;
        case DUCKDB_TYPE_USMALLINT:  return LogicalType::USMALLINT;
        case DUCKDB_TYPE_UINTEGER:   return LogicalType::UINTEGER;
        case DUCKDB_TYPE_UBIGINT:    return LogicalType::UBIGINT;
        case DUCKDB_TYPE_FLOAT:      return LogicalType::FLOAT;
        case DUCKDB_TYPE_DOUBLE:     return LogicalType::DOUBLE;
        case DUCKDB_TYPE_TIMESTAMP:  return LogicalType::TIMESTAMP;
        case DUCKDB_TYPE_DATE:       return LogicalType::DATE;
        case DUCKDB_TYPE_TIME:       return LogicalType::TIME;
        case DUCKDB_TYPE_INTERVAL:   return LogicalType::INTERVAL;
        case DUCKDB_TYPE_HUGEINT:    return LogicalType::BIGINT;   // normalised
        case DUCKDB_TYPE_UHUGEINT:   return LogicalType::UBIGINT;  // normalised
        case DUCKDB_TYPE_VARCHAR:    return LogicalType::VARCHAR;
        case DUCKDB_TYPE_BLOB:       return LogicalType::BLOB;
        case DUCKDB_TYPE_TIMESTAMP_S:  return LogicalType::TIMESTAMP_S;
        case DUCKDB_TYPE_TIMESTAMP_MS: return LogicalType::TIMESTAMP_MS;
        case DUCKDB_TYPE_TIMESTAMP_NS: return LogicalType::TIMESTAMP_NS;
        case DUCKDB_TYPE_TIMESTAMP_TZ: return LogicalType::TIMESTAMP_TZ;
        case DUCKDB_TYPE_TIME_TZ:    return LogicalType::TIME_TZ;
        case DUCKDB_TYPE_UUID:       return LogicalType::UUID;
        case DUCKDB_TYPE_BIT:        return LogicalType::BIT;
        case DUCKDB_TYPE_ENUM:       return LogicalType::VARCHAR;  // declared as VARCHAR
        case DUCKDB_TYPE_STRUCT:     return LogicalType::VARCHAR;  // fallback to VARCHAR
        case DUCKDB_TYPE_MAP:        return LogicalType::VARCHAR;
        case DUCKDB_TYPE_DECIMAL:    return LogicalType::DECIMAL(18, 3);  // placeholder
        case DUCKDB_TYPE_LIST:       return LogicalType::LIST(LogicalType::VARCHAR);  // placeholder
        default:
            // Unknown type — declare VARCHAR so downstream operators don't
            // get a junk LogicalTypeId. Type mismatch will surface at exec.
            return LogicalType::VARCHAR;
    }
}

struct SemanticViewBindData : public TableFunctionData {
    std::vector<SemanticViewColumnInfo> columns;
    std::string execution_sql;
    std::string expanded_sql_for_error;  // Mirror of execution_sql for SqlExecution error.
};

struct SemanticViewGlobalState : public GlobalTableFunctionState {
    // `result` is drained by a single, unsynchronized cursor: the exec
    // function (`sv_semantic_view_function`) calls `result->Fetch()` with no
    // lock. That is safe ONLY because this state does not override
    // `MaxThreads()`, so it inherits the base `GlobalTableFunctionState`
    // default of 1 — DuckDB scans the semantic_view() table function
    // single-threaded and never calls the exec function concurrently on the
    // same global state. If a future change overrides `MaxThreads()` to allow
    // parallel scans, the `Fetch()` cursor must be guarded (mutex, or a
    // per-thread local state) or it will data-race.
    std::unique_ptr<MaterializedQueryResult> result;
};

// Run `sql` on `probe` the way `Connection::Query` would, but polling the OUTER
// caller's interrupt flag so a cancelled query actually stops.
//
// `semantic_view()` executes its expanded SQL on a per-call
// `Connection(*context.db)`, which carries its OWN `ClientContext`, and DuckDB's
// interrupt flag is per-`ClientContext`. `duckdb_interrupt()` — and therefore
// ADBC's `adbc_cancel()`, the CLI's Ctrl-C, and any framework timeout built on
// them — sets `interrupted` on the *caller's* context. A plain
// `probe.Query(sql)` blocks until the inner query finishes and only ever reads
// the inner context's flag, which nothing sets, so a cancelled query ran to full
// completion and the INTERRUPT surfaced only afterwards. Nor is there a yield
// point in between: the inner query runs eagerly from a single outer executor
// task, and the outer executor's own interrupt checks live in
// `PipelineExecutor::Execute`, which does not run until this callback returns.
//
// So this unrolls the loop `Connection::Query` runs internally
// (`PendingQueryResult::ExecuteInternal`) to get a poll point between tasks. On a
// cancel the flag is forwarded to the inner connection and the loop keeps
// draining rather than throwing immediately, so the inner executor's tasks are
// wound down before `probe` leaves the caller's scope.
//
// The interrupt MUST be reported bare. ADBC maps a failure to
// ADBC_STATUS_CANCELLED only on an exact match against
// `InterruptException::INTERRUPT_MESSAGE` ("Interrupted!"), so wrapping it the
// way the SqlExecution path wraps every other failure would report a cancelled
// query as an internal error. Hence it is thrown from in here, ahead of any
// caller's wrap, instead of being returned as an error-carrying result.
//
// Interrupt latency is one task slice. A BLOCKED inner query (async I/O — httpfs
// and friends) adds at most one `Executor::WAIT_TIME` (20 ms) per poll, since
// `WaitForTask()` waits with that timeout rather than indefinitely.
//
// Everything else matches `Connection::Query`, including returning an
// error-carrying result rather than throwing, so callers keep their existing
// `HasError()` handling. The one behavioural difference: a *multi*-statement
// string would be rejected ("Cannot prepare multiple statements at once!")
// instead of executed. `execution_sql` is always a single generated SELECT.
static unique_ptr<QueryResult> sv_query_interruptible(Connection &probe,
                                                      ClientContext &outer,
                                                      const std::string &sql) {
    auto pending = probe.PendingQuery(sql);
    if (pending->HasError()) {
        // Parse/bind failure — nothing has executed. Same shape
        // Connection::Query hands back for the identical failure.
        return make_uniq<MaterializedQueryResult>(pending->GetErrorObject());
    }

    bool forwarded = false;
    PendingExecutionResult exec_result = PendingExecutionResult::RESULT_NOT_READY;
    while (!PendingQueryResult::IsResultReady(exec_result)) {
        if (!forwarded && outer.interrupted) {
            // Sets the inner context's flag; the inner executor picks it up at
            // its own next poll and unwinds into an INTERRUPT error, which ends
            // this loop via EXECUTION_ERROR.
            forwarded = true;
            probe.Interrupt();
        }
        exec_result = pending->ExecuteTask();
        if (exec_result == PendingExecutionResult::BLOCKED) {
            pending->WaitForTask();
        }
    }

    if (outer.interrupted) {
        // The caller cancelled — authoritative however the inner query ended.
        // It may have raced to completion, or stopped on the forwarded flag.
        throw InterruptException();
    }
    if (pending->HasError()) {
        return make_uniq<MaterializedQueryResult>(pending->GetErrorObject());
    }
    return pending->Execute();
}

// Look up DECIMAL/LIST logical types via a LIMIT-0 probe. Returns a
// per-column LogicalType for every column in the declared schema; for
// columns whose type_id isn't DECIMAL/LIST, the LogicalType is constructed
// from the type_id directly (matching the behaviour of
// `type_from_duckdb_type_u32` on the Rust side).
//
// Phase 65.1 WR-03: ENUM is intentionally declared as VARCHAR via
// `sv_logical_type_from_c_type_id` (see that helper) — the LIMIT-0 probe
// is NOT required for ENUM. The earlier comment claiming otherwise was a
// maintenance trap.
//
// Phase 65.1 WR-07: takes the caller's per-bind Connection by reference
// rather than opening a fresh one. This eliminates one ctor/dtor pair
// per `semantic_view(...)` invocation when DECIMAL/LIST is present and,
// more importantly, ensures the bind-time LIMIT-0 query runs on the same
// Connection (and therefore the same transaction/catalog snapshot) as
// the Rust dispatcher's catalog reads. Previously the FFI dispatcher
// borrowed one Connection while this helper opened a second — two
// separate Connections viewing potentially different uncommitted state.
//
// Probe failures (DuckDB error from the LIMIT 0 query) and column-count
// mismatches now raise BinderException with full diagnostic text instead
// of silently falling back to a type_id-only declaration (Phase 65.1
// Plan 11 / WR-07 / D-14). The previous fallback masked real DDL-level
// errors (broken FACTS / METRICS expressions referencing nonexistent
// columns) behind a DECIMAL(18,3) placeholder at query time.
static std::vector<LogicalType> sv_resolve_output_logical_types(
    Connection &probe,
    const std::vector<SemanticViewColumnInfo> &cols,
    const std::string &execution_sql) {
    std::vector<LogicalType> out;
    out.reserve(cols.size());

    bool needs_logical_probe = false;
    for (const auto &c : cols) {
        // C-API DUCKDB_TYPE_* enum values from duckdb.h. These two need
        // logical-type metadata to declare correctly: DECIMAL needs
        // width+scale; LIST needs child type. ENUM is intentionally
        // declared as VARCHAR via sv_logical_type_from_c_type_id, so no
        // probe is needed for it (Phase 65.1 WR-03).
        if (c.type_id == DUCKDB_TYPE_DECIMAL ||
            c.type_id == DUCKDB_TYPE_LIST) {
            needs_logical_probe = true;
            break;
        }
    }

    if (!needs_logical_probe) {
        // Fast path: build each LogicalType straight from the C-API type_id.
        for (const auto &c : cols) {
            out.emplace_back(sv_logical_type_from_c_type_id(c.type_id));
        }
        return out;
    }

    // Slow path: run LIMIT 0 on the caller-supplied Connection to extract
    // logical type metadata for DECIMAL/LIST columns. Reusing the bind's
    // Connection (rather than opening a fresh one) preserves the
    // transaction/catalog snapshot the FFI dispatcher already borrowed.
    std::string limit0_sql = "SELECT * FROM (" + execution_sql + ") __sv_probe LIMIT 0";
    auto probe_result = probe.Query(limit0_sql);
    if (probe_result->HasError()) {
        // Phase 65.1 Plan 11 / WR-07 / D-14: surface the underlying DuckDB
        // error verbatim. No silent fallback to a type_id-only declaration.
        throw BinderException(
            "semantic_view: failed to infer column types via LIMIT 0 probe: " +
            probe_result->GetError());
    }
    if (probe_result->types.size() != cols.size()) {
        // Phase 65.1 Plan 11 / WR-07 / D-14: distinct diagnostic naming
        // both counts so a future contributor can see which side disagrees.
        throw BinderException(
            "semantic_view: LIMIT 0 probe returned " +
            std::to_string(probe_result->types.size()) +
            " columns, expected " + std::to_string(cols.size()));
    }

    for (idx_t i = 0; i < cols.size(); ++i) {
        const auto &c = cols[i];
        const LogicalType &probed = probe_result->types[i];
        if (c.type_id == DUCKDB_TYPE_DECIMAL || c.type_id == DUCKDB_TYPE_LIST) {
            // Use the probed logical type (preserves width/scale/child type).
            out.emplace_back(probed);
        } else {
            out.emplace_back(sv_logical_type_from_c_type_id(c.type_id));
        }
    }
    return out;
}

static unique_ptr<FunctionData> sv_semantic_view_bind(
    ClientContext &context,
    TableFunctionBindInput &input,
    vector<LogicalType> &return_types,
    vector<string> &names) {
    if (input.inputs.empty() || input.inputs[0].IsNull()) {
        throw BinderException(
            "semantic_view: view name is required (positional arg 0)");
    }
    std::string view_name = input.inputs[0].GetValue<std::string>();

    std::vector<uint8_t> dims_buf, metrics_buf, facts_buf;
    auto it_d = input.named_parameters.find("dimensions");
    if (it_d != input.named_parameters.end() && !it_d->second.IsNull()) {
        dims_buf = sv_serialise_string_list(it_d->second, "dimensions");
    }
    auto it_m = input.named_parameters.find("metrics");
    if (it_m != input.named_parameters.end() && !it_m->second.IsNull()) {
        metrics_buf = sv_serialise_string_list(it_m->second, "metrics");
    }
    auto it_f = input.named_parameters.find("facts");
    if (it_f != input.named_parameters.end() && !it_f->second.IsNull()) {
        facts_buf = sv_serialise_string_list(it_f->second, "facts");
    }

    // Scalar VARCHAR, not a list — passed to Rust as a plain UTF-8 ptr/len.
    // Absent or NULL yields (nullptr, 0), which the Rust side reads as "no
    // predicate".
    std::string where_clause;
    auto it_w = input.named_parameters.find("where_clause");
    if (it_w != input.named_parameters.end() && !it_w->second.IsNull()) {
        where_clause = it_w->second.GetValue<std::string>();
    }
    std::vector<uint8_t> sp_buf = sv_search_path_payload(input);

    Connection probe(*context.db);
    duckdb_connection borrowed = reinterpret_cast<duckdb_connection>(&probe);

    SvOwnedBuffer payload;
    char error_buf[1024];
    std::memset(error_buf, 0, sizeof(error_buf));
    uint8_t rc = sv_semantic_view_bind_rust(
        borrowed,
        reinterpret_cast<const uint8_t *>(view_name.data()), view_name.size(),
        sp_buf.empty()      ? nullptr : sp_buf.data(),      sp_buf.size(),
        dims_buf.empty()    ? nullptr : dims_buf.data(),    dims_buf.size(),
        metrics_buf.empty() ? nullptr : metrics_buf.data(), metrics_buf.size(),
        facts_buf.empty()   ? nullptr : facts_buf.data(),   facts_buf.size(),
        where_clause.empty() ? nullptr : reinterpret_cast<const uint8_t *>(where_clause.data()),
        where_clause.size(),
        &payload.ptr, &payload.len,
        error_buf, sizeof(error_buf));
    if (rc != 0) {
        throw BinderException(std::string("semantic_view: ") + error_buf);
    }

    auto bd = make_uniq<SemanticViewBindData>();

    // Parse the schema + execution_sql wire format.
    size_t offset = 0;
    uint32_t n_cols = sv_read_u32_le(payload.ptr, payload.len, offset, "semantic_view");
    bd->columns.reserve(n_cols);
    for (uint32_t i = 0; i < n_cols; ++i) {
        SemanticViewColumnInfo info;
        info.name = sv_read_string(payload.ptr, payload.len, offset, "semantic_view");
        info.type_id = sv_read_u32_le(payload.ptr, payload.len, offset, "semantic_view");
        bd->columns.push_back(std::move(info));
    }
    bd->execution_sql = sv_read_string(payload.ptr, payload.len, offset, "semantic_view");
    if (offset != payload.len) {
        throw BinderException(
            "semantic_view: FFI buffer has trailing bytes (consumed " +
            std::to_string(offset) + " of " + std::to_string(payload.len) + ")");
    }
    bd->expanded_sql_for_error = bd->execution_sql;

    // Resolve declared logical types — runs a LIMIT-0 probe on the SAME
    // Connection the FFI dispatcher already borrowed, if any DECIMAL/LIST
    // column is in the schema (so width/scale/child-type can be honoured).
    // Phase 65.1 WR-07: reusing `probe` here avoids a second Connection
    // ctor/dtor pair and keeps both queries on the same
    // transaction/catalog snapshot.
    auto declared_types = sv_resolve_output_logical_types(
        probe, bd->columns, bd->execution_sql);
    for (idx_t i = 0; i < bd->columns.size(); ++i) {
        return_types.push_back(declared_types[i]);
        names.push_back(bd->columns[i].name);
    }
    return std::move(bd);
}

static unique_ptr<GlobalTableFunctionState> sv_semantic_view_init_global(
    ClientContext &context,
    TableFunctionInitInput &input) {
    auto &bd = input.bind_data->Cast<SemanticViewBindData>();
    auto state = make_uniq<SemanticViewGlobalState>();

    // Open a per-call Connection on the caller's DatabaseInstance and run
    // the materialised query. The Connection drops at end of scope; the
    // returned MaterializedQueryResult owns its data (via ColumnDataCollection)
    // and is safe to use across exec invocations.
    //
    // Routed through sv_query_interruptible rather than Connection::Query so a
    // cancel on the CALLER's connection stops this query instead of being
    // noticed only once it has finished — the inner query runs on `probe`'s own
    // ClientContext, which carries its own interrupt flag. An interrupt leaves
    // via a bare InterruptException thrown from in there, deliberately ahead of
    // the SqlExecution wrap below: ADBC recognises a cancellation only by the
    // exact "Interrupted!" text, so wrapping it would report it as an internal
    // error.
    Connection probe(*context.db);
    auto qresult = sv_query_interruptible(probe, context, bd.execution_sql);
    if (qresult->HasError()) {
        // Match the legacy QueryError::SqlExecution wording so existing
        // sqllogictest matchers stay byte-identical.
        throw InvalidInputException(
            "semantic_view: SQL execution failed: " + qresult->GetError() +
            "  (expanded SQL: " + bd.expanded_sql_for_error + ")");
    }
    if (qresult->type != QueryResultType::MATERIALIZED_RESULT) {
        throw InternalException(
            "semantic_view: expected MaterializedQueryResult from Connection::Query");
    }
    state->result = unique_ptr_cast<QueryResult, MaterializedQueryResult>(std::move(qresult));
    return std::move(state);
}

static void sv_semantic_view_function(
    ClientContext & /*context*/,
    TableFunctionInput &data_p,
    DataChunk &output) {
    auto &bd = data_p.bind_data->Cast<SemanticViewBindData>();
    auto &gs = data_p.global_state->Cast<SemanticViewGlobalState>();

    // Fetch the next chunk from the materialised result. Fetch() flattens
    // vectors and returns nullptr when the stream is exhausted.
    auto chunk = gs.result->Fetch();
    if (!chunk || chunk->size() == 0) {
        output.SetCardinality(0);
        return;
    }

    // Validate column count — defensive against a planner inserting projections
    // we didn't anticipate. The legacy Rust impl had a Mutex + col_count check;
    // for the C++ path the bind-time schema agreement is authoritative.
    //
    // Phase 65.1 WR-02: previously this took `std::min(chunk->ColumnCount(),
    // bd.columns.size())` and silently truncated on mismatch, leaving any
    // unfilled output slots with default-constructed Vector contents that
    // are not guaranteed to be NULL-marked — readers downstream could see
    // garbage. Replace with an explicit mismatch error so a future planner
    // change that violates the bind contract fails loud at first emission
    // rather than producing silently-wrong data.
    if (chunk->ColumnCount() != bd.columns.size()) {
        throw InternalException(
            "semantic_view: column count mismatch — chunk has " +
            std::to_string(chunk->ColumnCount()) + " columns, bind declared " +
            std::to_string(bd.columns.size()) +
            " (please report this bug)");
    }
    idx_t n_cols = chunk->ColumnCount();

    // Copy each column into the output chunk via Vector::Reference (zero-copy
    // when types match exactly; falls back to a cast when build_execution_sql
    // already inserted a `::TYPE` wrapper but the runtime type still differs).
    for (idx_t col_idx = 0; col_idx < n_cols; ++col_idx) {
        auto &src = chunk->data[col_idx];
        auto &dst = output.data[col_idx];
        if (src.GetType() == dst.GetType()) {
            // Same logical type — zero-copy reference.
            dst.Reference(src);
        } else {
            // Type mismatch despite the bind-time cast wrapper — last-resort
            // cast. Matches the legacy QueryError::TypeMismatch fallback but
            // does the cast instead of erroring; the bind-time cast wrapper
            // (build_execution_sql) is the primary defence.
            VectorOperations::DefaultCast(src, dst, chunk->size());
        }
    }
    output.SetCardinality(chunk->size());
}

// semantic_view takes the same named LIST(VARCHAR) parameters as
// explain_semantic_view, but uses an init_global state constructor (its exec
// streams the materialised query result across invocations) rather than
// init_local. It builds a spec from the single shared `sv_semantic_named_params`
// and registers via `sv_register_table_function_core` (C-7); the shared triple
// makes it impossible for this TF and explain_semantic_view to drift. Same
// per-call borrow contract and error_buf convention (D-08/D-09); no stderr.
static bool sv_register_semantic_view_impl(duckdb_database db_handle,
                                           char *error_buf,
                                           size_t error_buf_len) {
    const LogicalType arg_types[] = {LogicalType::VARCHAR};
    SvTableFunctionSpec spec;
    spec.name = "semantic_view";
    spec.arg_types = arg_types;
    spec.arg_count = 1;
    spec.named_params = sv_semantic_named_params();
    spec.bind_cb = sv_semantic_view_bind;
    spec.exec_cb = sv_semantic_view_function;
    spec.init_local_cb = nullptr;
    spec.init_global_cb = sv_semantic_view_init_global;
    return sv_register_table_function_core(
        db_handle, spec, "sv_register_semantic_view", error_buf,
        error_buf_len);
}

extern "C" {
    bool sv_register_semantic_view(duckdb_database db_handle,
                                   char *error_buf, size_t error_buf_len) {
        return sv_register_semantic_view_impl(
            db_handle, error_buf, error_buf_len);
    }
}

// ---------------------------------------------------------------------------
// sv_register_parser_hooks -- called from Rust after C API init
// ---------------------------------------------------------------------------
// Extracts DatabaseInstance& from the C API handle and registers the
// parser_override hook on DBConfig. Phase 65 Plan 06: signature slimmed
// to `(db_handle)` after H1 catalog_conn retirement. AR-7: the
// SemanticViewsParserInfo carries no Rust state anymore — it is just the
// dynamic_cast marker used by sv_parser_override / sv_parse_stub to confirm
// the parser_info is ours.
//
// Phase 65.1 Plan 10 (WR-06 D-12/D-13): the signature now carries the
// trailing `(char *error_buf, size_t error_buf_len)` pair — matching the
// 17 read-side dispatchers — so registration failures and BORROW-contract
// probe failures surface through the ABI-stable channel into the Rust
// caller's diagnostic. A runtime probe right after the wrapper unwrap
// catches representation drift the file-scope `static_assert` cannot.
extern "C" {
    bool sv_register_parser_hooks(
        duckdb_database db_handle,
        char *error_buf, size_t error_buf_len) {
        try {
            // duckdb_database -> internal_ptr -> DatabaseWrapper ->
            //   shared_ptr<DuckDB> -> shared_ptr<DatabaseInstance>
            auto *wrapper = reinterpret_cast<duckdb::DatabaseWrapper *>(
                db_handle->internal_ptr);
            auto &db = *wrapper->database->instance;

            // Phase 65.1 Plan 10 (WR-06 D-12 runtime probe). The file-scope
            // `static_assert(sizeof(duckdb_connection) == sizeof(void*))`
            // catches size drift; this probe catches representation drift
            // that an equal-size change would miss (e.g. wrapping the
            // pointer in a struct with the same layout but different
            // semantics). Failure refuses the LOAD entirely so no
            // semantic_views state survives a broken bridge — better than
            // discovering the breakage on the first user query.
            //
            // The `Connection probe` is stack-allocated; its dtor runs at
            // scope exit. The bridged `handle` becomes invalid but is not
            // touched after the SELECT 1 round-trip.
            {
                Connection probe(db);
                auto handle = reinterpret_cast<duckdb_connection>(&probe);
                duckdb_result r;
                auto rc = duckdb_query(handle, "SELECT 1", &r);
                if (rc != DuckDBSuccess) {
                    // Phase 65.1 IN-01: surface the underlying DuckDB error
                    // text and clarify that the probe failure can stem from
                    // either ABI drift (the original BORROW-contract concern)
                    // OR a transient DB error at LOAD time (shutdown in
                    // progress, OOM, etc.). The diagnostic must list both
                    // possible causes so an operator hitting a non-bridge
                    // failure isn't misled into chasing an ABI regression.
                    const char *err_text = duckdb_result_error(&r);
                    if (error_buf != nullptr && error_buf_len > 0) {
                        snprintf(error_buf, error_buf_len,
                            "bridge contract probe failed (duckdb_query "
                            "SELECT 1 returned %d): %s. Possible causes: "
                            "duckdb_connection ABI drift OR transient DB "
                            "error at LOAD time; refusing to load "
                            "semantic_views extension.",
                            (int)rc, err_text ? err_text : "(no detail)");
                    }
                    duckdb_destroy_result(&r);
                    return false;
                }
                duckdb_destroy_result(&r);
            }

            // Phase 65 Plan 04 (Task 2 Step B): register the
            // `__sv_compute_create_from_yaml` helper TF via the C++ Catalog
            // API so its bind callback receives a native `ClientContext &`
            // and can open per-call `Connection(*context.db)` to read the
            // YAML file. The outer parser_override INSERT in
            // src/parse.rs::rewrite_yaml_file_create wraps the helper TF's
            // `new_def` row with json_merge_patch + json_object to add
            // now()/current_database()/current_schema() on the caller's
            // connection, preserving D-21 transactional contract.
            //
            // LIFE-2 (code-review 2026-08-08): this block runs BEFORE the
            // parser-extension registration below, for the same reason
            // `REGISTRATIONS` in src/lib.rs puts the whole hook last — the
            // dependency is one-way. The helper TF is a pure Catalog-API
            // registration that never needs the hook, while the hook's own
            // rewrite of `CREATE SEMANTIC VIEW ... FROM YAML FILE` emits a
            // call to this TF. Registering the hook first meant a failure
            // here returned false with `parser_override` already installed
            // and `allow_parser_override_extension` already flipped to
            // FALLBACK, so the YAML-file DDL form parsed and then failed on
            // an unresolved function name.
            {
                // Phase 65.1 Plan 07 (IN-04 D-24): helper TF slimmed to
                // 3 args — (file_path, view_name, comment). The previous
                // `kind` INTEGER parameter was redundant with the outer
                // parser_override INSERT shape (OR IGNORE / OR REPLACE /
                // plain) and the Rust helper ignored it.
                LogicalType arg_types[] = {
                    LogicalType::VARCHAR,  // file_path
                    LogicalType::VARCHAR,  // view_name
                    LogicalType::VARCHAR,  // comment (empty = none)
                };
                // Phase 65.1 Plan 02a: sv_register_table_function requires a
                // `(error_buf, error_buf_len)` trailing pair.
                //
                // Phase 65.1 Plan 10 (WR-06 D-13): sv_register_parser_hooks
                // now carries its own error_buf; forward the helper TF's
                // diagnostic into the parent buffer so the Rust caller
                // surfaces it through `decode_register_err_buf`. The local
                // buffer is needed because the helper writes via snprintf
                // and may truncate independently of our outer buffer's size.
                char helper_err[1024];
                std::memset(helper_err, 0, sizeof(helper_err));
                if (!sv_register_table_function(
                        db_handle,
                        "__sv_compute_create_from_yaml",
                        arg_types, 3,
                        sv_create_from_yaml_bind,
                        sv_create_from_yaml_function,
                        sv_create_from_yaml_init_local,
                        helper_err, sizeof(helper_err))) {
                    if (error_buf != nullptr && error_buf_len > 0) {
                        snprintf(error_buf, error_buf_len,
                            "sv_register_parser_hooks: failed to register "
                            "__sv_compute_create_from_yaml helper TF: %s",
                            helper_err);
                    }
                    return false;
                }
            }

            auto &config = DBConfig::GetConfig(db);

            // Phase 65.1 Plan 12 (WR-09 D-21): idempotence check across
            // same-process repeat LOAD semantic_views. `ParserExtension::Register`
            // unconditionally appends to `DBConfig::parser_extensions` (no dedup
            // API), so without this guard every LOAD grows the list by one entry
            // pointing at the same `sv_parser_override` — an unbounded "soft leak"
            // surfaced by 65-REVIEW.md WR-09.
            //
            // We iterate the public read-side iterator
            // `DBConfig::GetCallbackManager().ParserExtensions()`
            // (cpp/include/duckdb.cpp:281157) and compare each existing entry's
            // `parser_override` function pointer against the file-static
            // `sv_parser_override` symbol. Function-pointer equality is portable
            // ISO C++ here because `sv_parser_override` is a file-static C++
            // function defined in this same TU — both sides of the `==` see the
            // identical pointer value.
            //
            // CRITICAL: the entire `ParserExtension ext; ...; Register(...)` block
            // — including the `SemanticViewsParserInfo` allocation — must be
            // guarded. `Register` unconditionally appends to
            // `DBConfig::parser_extensions`; on the skip path re-registering
            // would grow the list unboundedly across repeated re-LOAD cycles
            // (the precise leak shape WR-09 / CR-01 (65.1) was meant to fix).
            //
            // The target is same-process repeat LOAD (e.g. a long-lived host
            // re-invoking `LOAD semantic_views` across sessions, or partial-failure
            // retries during `init_extension`). The theoretical concurrent-thread
            // LOAD race between the iterator check and `Register` is documented
            // as out of scope for v0.10.0 — see WR-09 Pitfall 3 / RESEARCH.md
            // lines 668-714. Hitting it would require user code to LOAD the
            // extension from two threads simultaneously, which is atypical for
            // DuckDB's extension-load API.
            //
            // The helper-TF registration above uses
            // `OnCreateConflict::ALTER_ON_CONFLICT` and is naturally
            // idempotent — no analog guard needed there.
            auto &cbmgr = config.GetCallbackManager();
            bool already_registered = false;
            for (auto &existing : cbmgr.ParserExtensions()) {
                if (existing.parser_override == sv_parser_override) {
                    already_registered = true;
                    break;
                }
            }

            if (!already_registered) {
                // Phase 65.1 CR-01 / WR-09: the `ParserExtension ext; ...;
                // Register(...)` block — including the `SemanticViewsParserInfo`
                // allocation — must stay INSIDE the dedup guard.
                // `ParserExtension::Register` unconditionally appends to
                // `DBConfig::parser_extensions` (no dedup API), so on the
                // `already_registered` skip path we must not Register (or
                // allocate) again — otherwise every redundant LOAD grows the
                // list by one entry, an unbounded soft leak over a long-lived
                // host process. AR-7: there is no longer a separate
                // `sv_make_override_context()` allocation to guard here.
                ParserExtension ext;
                ext.parser_override = sv_parser_override;
                // Phase 62 Plan 03: parse_function is the error-reporting layer.
                // parser_override owns the success path (rewrite + re-parse on the
                // caller's connection — transactional). When parser_override
                // defers (rc=2), the default parser fails on the unrecognised
                // prefix and DuckDB calls sv_parse_stub, which re-runs validation
                // and returns DISPLAY_EXTENSION_ERROR with `error_location` so
                // ParserException::SyntaxError renders `LINE 1: … ^` (caret).
                // sv_plan_unreachable is the required sibling — sv_parse_stub
                // never returns PARSE_SUCCESSFUL so it should never fire.
                ext.parse_function  = sv_parse_stub;
                ext.plan_function   = sv_plan_unreachable;
                // duckdb::shared_ptr<T> doesn't have an upcast constructor, so we
                // build the std::shared_ptr<ParserExtensionInfo> first (allocates
                // the SemanticViewsParserInfo and immediately upcasts) then hand
                // it to duckdb::shared_ptr's std-interop constructor.
                std::shared_ptr<ParserExtensionInfo> info_std(
                    new SemanticViewsParserInfo());
                ext.parser_info = duckdb::shared_ptr<ParserExtensionInfo>(info_std);
                ParserExtension::Register(config, ext);

                // FALLBACK_OVERRIDE: hook runs before default parser; misses
                // fall through cleanly. Caret regression (TECH-DEBT 22) is
                // resolved by parse_function above — sv_parse_stub renders
                // `LINE 1: ... ^` for every CREATE/DROP/ALTER validation error.
                config.SetOption("allow_parser_override_extension", Value("FALLBACK"));
            }

            return true;
        } catch (const std::exception &e) {
            if (error_buf != nullptr && error_buf_len > 0) {
                snprintf(error_buf, error_buf_len,
                    "sv_register_parser_hooks failed: %s", e.what());
            }
            return false;
        }
    }
}

// ---------------------------------------------------------------------------
// sv_count_parser_extensions -- Phase 65.1 Plan 12 Task 3 (D-21 + B-07
// plan-checker fix) structural verification helper.
// ---------------------------------------------------------------------------
// Counts entries in `DBConfig::GetCallbackManager().ParserExtensions()`
// whose `parser_override` function pointer equals the file-static
// `sv_parser_override` symbol. Used by
// `tests/parser_hook_idempotent.rs` as a stable, named symbol the
// structural test can reference (the cargo-test FFI path cannot
// exercise the extension binary directly under the
// `duckdb/loadable-extension` stubs; see that test's docstring).
//
// Option (a) per the B-07 plan-checker fix in 65.1-12-PLAN.md: PUBLIC
// helper (no cfg(test) gating). Read-only iteration is harmless in
// production binaries and gating would require conditional-compilation
// plumbing in the C++ TU for marginal benefit.
//
// Returns >= 0 on success (the count); -1 on failure with the message
// in `error_buf`. Same error_buf convention as the WR-02 ABI.
extern "C" {
    int32_t sv_count_parser_extensions(
        duckdb_database db_handle,
        char *error_buf, size_t error_buf_len) {
        try {
            if (db_handle == nullptr) {
                if (error_buf != nullptr && error_buf_len > 0) {
                    snprintf(error_buf, error_buf_len,
                        "sv_count_parser_extensions: null db_handle");
                }
                return -1;
            }
            auto *wrapper = reinterpret_cast<duckdb::DatabaseWrapper *>(
                db_handle->internal_ptr);
            if (wrapper == nullptr) {
                if (error_buf != nullptr && error_buf_len > 0) {
                    snprintf(error_buf, error_buf_len,
                        "sv_count_parser_extensions: null DatabaseWrapper");
                }
                return -1;
            }
            auto &db = *wrapper->database->instance;
            auto &config = DBConfig::GetConfig(db);
            auto &cbmgr = config.GetCallbackManager();
            // Phase 65.1 IN-02: saturate the counter at INT32_MAX rather
            // than wrapping. The documented contract is "returns >= 0 on
            // success; -1 on failure" — silent wrap to a negative value
            // would be misinterpreted as a registration failure by the
            // structural test. Reaching INT32_MAX is only plausible if
            // WR-09's dedup guard fails badly and the list grows
            // unbounded across re-LOAD cycles, but the saturation is
            // cheap defence-in-depth.
            int32_t count = 0;
            for (auto &existing : cbmgr.ParserExtensions()) {
                if (existing.parser_override == sv_parser_override) {
                    if (count < INT32_MAX) {
                        ++count;
                    }
                }
            }
            return count;
        } catch (const std::exception &e) {
            if (error_buf != nullptr && error_buf_len > 0) {
                snprintf(error_buf, error_buf_len,
                    "sv_count_parser_extensions failed: %s", e.what());
            }
            return -1;
        } catch (...) {
            if (error_buf != nullptr && error_buf_len > 0) {
                snprintf(error_buf, error_buf_len,
                    "sv_count_parser_extensions failed: unknown C++ exception");
            }
            return -1;
        }
    }
}
