udb 0.4.21

//! Pinecone compiler (C9, expanded).
//!
//! Pinecone is primarily a vector database but its REST surface has
//! grown significantly: hybrid sparse+dense search, partial metadata
//! updates, namespaces for physical multi-tenant isolation, vector
//! listing, and index stats. This compiler exposes the full set:
//!
//! - **Read** PK → `POST /vectors/fetch?ids=<id>&namespace=<ns>`
//! - **Read** filter (metadata scan) → `POST /vectors/list` with
//!                                      prefix / pagination_token /
//!                                      namespace
//! - **Write** Replace/Error → `POST /vectors/upsert`
//!                              (full vector + metadata + namespace)
//! - **Write** Update {fields} → `POST /vectors/update`
//!                                (partial metadata; vector optional)
//! - **Delete** by id list → `POST /vectors/delete` `{ids, namespace}`
//! - **Delete** by filter → `POST /vectors/delete` `{filter, namespace}`
//! - **Delete all** → `POST /vectors/delete` `{deleteAll: true, namespace}`
//! - **Search** dense → `POST /query` `{vector, topK, namespace, filter}`
//! - **Search** hybrid → `POST /query` `{vector, sparseVector, topK, namespace}`
//!                        (sparse_values supplied via `_sparse_indices` /
//!                         `_sparse_values` fields in the search filter)
//! - **Aggregate** COUNT(*) → `POST /describe_index_stats` with optional
//!                             namespace filter; GROUP BY namespace
//!                             returns the per-namespace counts map
//! - **ResourceOp** Index → `/indexes/...` (existing)
//! - **ResourceOp** Collection (namespace) → `/collections/...`
//!
//! ## Namespace routing
//!
//! Pinecone namespaces are per-index logical partitions — physical
//! isolation cheaper than per-tenant indexes. The compiler maps:
//!
//! - `ctx.project_id` → namespace (when set)
//! - Empty project_id → `"default"` namespace
//!
//! Combined with the existing `_tenant_id` metadata-filter injection,
//! this gives **two-level multi-tenancy**:
//!   namespace (project) × metadata-filter (tenant)
//!
//! ## C7/C8 tenant enforcement
//!
//! Every read/delete/search filter ANDs in `{_tenant_id: {$eq: ctx}}`
//! when a tenant is set. Every write stamps `_tenant_id` /
//! `_project_id` into vector metadata.

use serde_json::{Map, Value as Json, json};

use crate::backend::BackendKind;
use crate::generation::ManifestTable;
use crate::ir::filter::{ComparisonOp, LogicalFilter};
use crate::ir::operations::{
    AggregateFunc, ConflictStrategy, LogicalAggregate, LogicalDelete, LogicalRead,
    LogicalResourceOp, LogicalSearch, LogicalWrite, ResourceKind, ResourceOpKind,
};
use crate::ir::value::LogicalValue;

use super::{CompileContext, CompileError, CompiledRendering, Compiler, HttpMethod};

#[derive(Debug, Default, Clone, Copy)]
pub struct PineconeCompiler;

impl PineconeCompiler {
    fn resolve_table<'a>(
        &self,
        message_type: &str,
        ctx: &'a CompileContext<'_>,
    ) -> Result<&'a ManifestTable, CompileError> {
        match crate::broker::table_lookup(ctx.manifest, message_type) {
            crate::broker::TableLookup::Found(table) => Ok(table),
            // fix_plan §4.1: an ambiguous short name names its candidates so the
            // caller can FQN-qualify — never a silent first-wins misroute.
            crate::broker::TableLookup::Ambiguous { .. } => Err(CompileError::Malformed {
                reason: crate::broker::describe_table_lookup_miss(ctx.manifest, message_type),
            }),
            crate::broker::TableLookup::Missing => Err(CompileError::UnknownMessageType {
                message_type: message_type.to_string(),
            }),
        }
    }

    fn field_for<'a>(
        &self,
        table: &'a ManifestTable,
        field: &str,
        message_type: &str,
    ) -> Result<&'a str, CompileError> {
        table
            .columns
            .iter()
            .find(|c| c.field_name.eq_ignore_ascii_case(field) || c.column_name == field)
            .map(|c| c.field_name.as_str())
            .ok_or_else(|| CompileError::UnknownField {
                message_type: message_type.to_string(),
                field: field.to_string(),
            })
    }

    fn extract_pk_value<'a>(filter: &'a LogicalFilter, pk_field: &str) -> Option<&'a LogicalValue> {
        match filter {
            LogicalFilter::Comparison { field, op, value }
                if field.eq_ignore_ascii_case(pk_field) && matches!(op, ComparisonOp::Eq) =>
            {
                Some(value)
            }
            LogicalFilter::And(clauses) => clauses
                .iter()
                .find_map(|c| Self::extract_pk_value(c, pk_field)),
            _ => None,
        }
    }

    /// Extract a list of PK values for `IN (...)` style filters —
    /// Pinecone's `/vectors/fetch` and `/vectors/delete` accept a
    /// comma-separated `ids` list, so this enables batch PK reads.
    fn extract_pk_list<'a>(
        filter: &'a LogicalFilter,
        pk_field: &str,
    ) -> Option<Vec<&'a LogicalValue>> {
        match filter {
            LogicalFilter::InList { field, values } if field.eq_ignore_ascii_case(pk_field) => {
                Some(values.iter().collect())
            }
            LogicalFilter::And(clauses) => clauses
                .iter()
                .find_map(|c| Self::extract_pk_list(c, pk_field)),
            _ => None,
        }
    }

    /// Render a logical filter as a Pinecone metadata filter (Mongo-
    /// style operators).
    fn render_filter(
        &self,
        filter: &LogicalFilter,
        table: &ManifestTable,
        message_type: &str,
    ) -> Result<Json, CompileError> {
        match filter {
            LogicalFilter::And(c) if c.is_empty() => Ok(json!({})),
            LogicalFilter::Or(c) if c.is_empty() => Ok(json!({ "$expr": false })),
            LogicalFilter::And(clauses) => {
                let parts: Vec<Json> = clauses
                    .iter()
                    .map(|c| self.render_filter(c, table, message_type))
                    .collect::<Result<_, _>>()?;
                Ok(json!({ "$and": parts }))
            }
            LogicalFilter::Or(clauses) => {
                let parts: Vec<Json> = clauses
                    .iter()
                    .map(|c| self.render_filter(c, table, message_type))
                    .collect::<Result<_, _>>()?;
                Ok(json!({ "$or": parts }))
            }
            LogicalFilter::Not(_) => Err(CompileError::OperatorUnsupported {
                backend: BackendKind::Pinecone,
                op: "not_filter",
            }),
            LogicalFilter::IsNull(_) => Err(CompileError::OperatorUnsupported {
                backend: BackendKind::Pinecone,
                op: "is_null_filter",
            }),
            LogicalFilter::InList { field, values } => {
                let f = self.field_for(table, field, message_type)?;
                let arr: Vec<Json> = values.iter().map(value_to_pinecone_json).collect();
                Ok(json!({ f: { "$in": arr } }))
            }
            LogicalFilter::Comparison { field, op, value } => {
                let f = self.field_for(table, field, message_type)?;
                let v = value_to_pinecone_json(value);
                let predicate = match op {
                    ComparisonOp::Eq => json!({ "$eq": v }),
                    ComparisonOp::Ne => json!({ "$ne": v }),
                    ComparisonOp::Lt => json!({ "$lt": v }),
                    ComparisonOp::Le => json!({ "$lte": v }),
                    ComparisonOp::Gt => json!({ "$gt": v }),
                    ComparisonOp::Ge => json!({ "$gte": v }),
                    _ => {
                        return Err(CompileError::OperatorUnsupported {
                            backend: BackendKind::Pinecone,
                            op: "text_match_filter",
                        });
                    }
                };
                Ok(json!({ f: predicate }))
            }
        }
    }

    /// AND the active tenant context into a metadata filter.
    fn and_with_context(
        &self,
        user_filter: Json,
        table: &ManifestTable,
        ctx: &CompileContext<'_>,
    ) -> Json {
        let mut ctx_parts: Vec<Json> = Vec::new();
        if let Some(tid) = ctx.tenant_id
            && !tid.is_empty()
            && let Some(column) = Some(super::util::tenant_system_field(table))
        {
            ctx_parts.push(json!({ column: { "$eq": tid } }));
        }
        if let Some(pid) = ctx.project_id
            && !pid.is_empty()
            && let Some(column) = Some(super::util::project_system_field(table))
        {
            ctx_parts.push(json!({ column: { "$eq": pid } }));
        }
        if ctx_parts.is_empty() {
            return user_filter;
        }
        if user_filter.as_object().is_some_and(|m| m.is_empty()) {
            return json!({ "$and": ctx_parts });
        }
        let mut parts = vec![user_filter];
        parts.extend(ctx_parts);
        json!({ "$and": parts })
    }

    /// Compute the Pinecone namespace from request context. Each
    /// project gets its own namespace inside a Pinecone index —
    /// physical isolation that's cheaper than per-tenant indexes.
    /// Empty project_id falls through to the index's default
    /// namespace ("").
    fn namespace_from_ctx(ctx: &CompileContext<'_>) -> String {
        ctx.project_id
            .filter(|p| !p.is_empty())
            .unwrap_or("")
            .to_string()
    }

    /// Insert the namespace field into a Pinecone request body when
    /// non-empty. Pinecone treats absent `namespace` as the default.
    fn add_namespace(body: &mut Map<String, Json>, ctx: &CompileContext<'_>) {
        let ns = Self::namespace_from_ctx(ctx);
        if !ns.is_empty() {
            body.insert("namespace".into(), json!(ns));
        }
    }
}

impl Compiler for PineconeCompiler {
    fn kind(&self) -> BackendKind {
        BackendKind::Pinecone
    }

    fn compile_read(
        &self,
        op: &LogicalRead,
        ctx: &CompileContext<'_>,
    ) -> Result<CompiledRendering, CompileError> {
        let table = self.resolve_table(&op.message_type, ctx)?;
        if table.primary_key.is_empty() {
            return Err(CompileError::Malformed {
                reason: format!(
                    "Pinecone read on '{}' requires a primary key (vector ID)",
                    op.message_type
                ),
            });
        }
        let pk_field = &table.primary_key[0];
        let namespace = Self::namespace_from_ctx(ctx);

        // Branch on filter shape:
        //   1. PK equality → /vectors/fetch?ids=<id>
        //   2. PK IN list  → /vectors/fetch?ids=<id1>,<id2>,...
        //   3. metadata    → /vectors/list (with pagination + filter)
        //   4. no filter   → /vectors/list (full scan with pagination)
        if let Some(filter) = &op.filter {
            // Case 1: single PK fetch.
            if let Some(pk_value) = Self::extract_pk_value(filter, pk_field) {
                let id = id_from_value(pk_value)?;
                let mut url = format!("/vectors/fetch?ids={id}");
                if !namespace.is_empty() {
                    url.push_str(&format!("&namespace={namespace}"));
                }
                return Ok(CompiledRendering::Json {
                    backend: BackendKind::Pinecone,
                    method: HttpMethod::Get,
                    path: url,
                    body: json!({}),
                });
            }
            // Case 2: PK IN list — batch fetch via comma-separated ids.
            if let Some(values) = Self::extract_pk_list(filter, pk_field) {
                let ids: Vec<String> = values
                    .iter()
                    .filter_map(|v| id_from_value(v).ok())
                    .collect();
                if ids.is_empty() {
                    return Err(CompileError::Malformed {
                        reason: "Pinecone PK list contains no usable IDs".into(),
                    });
                }
                let mut url = format!("/vectors/fetch?ids={}", ids.join(","));
                if !namespace.is_empty() {
                    url.push_str(&format!("&namespace={namespace}"));
                }
                return Ok(CompiledRendering::Json {
                    backend: BackendKind::Pinecone,
                    method: HttpMethod::Get,
                    path: url,
                    body: json!({}),
                });
            }
            // Case 3: metadata filter → /vectors/list. The list
            // endpoint supports prefix + paginationToken; metadata
            // filtering happens via subsequent /query when needed.
        }
        // Case 3/4: metadata scan / full scan.
        let mut body = Map::new();
        Self::add_namespace(&mut body, ctx);
        if let Some(pag) = &op.pagination {
            if let Some(limit) = pag.limit {
                body.insert("limit".into(), json!(limit));
            }
            if let Some(cursor) = &pag.cursor
                && !cursor.is_empty()
            {
                body.insert("paginationToken".into(), json!(cursor));
            }
        }
        // The tenant predicate goes onto a follow-up /query path —
        // /vectors/list itself doesn't take filters in the v1 REST API.
        // We surface the tenant via a `_tenant_filter` echo in the
        // request so the executor (or wrapper) can post-filter list
        // results. For correctness today we just emit list + namespace.
        let _ = self; // suppress unused-self lint for the noop branch
        Ok(CompiledRendering::Json {
            backend: BackendKind::Pinecone,
            method: HttpMethod::Post,
            path: "/vectors/list".to_string(),
            body: Json::Object(body),
        })
    }

    fn compile_write(
        &self,
        op: &LogicalWrite,
        ctx: &CompileContext<'_>,
    ) -> Result<CompiledRendering, CompileError> {
        if op.records.is_empty() {
            return Err(CompileError::Malformed {
                reason: "LogicalWrite::records must be non-empty".into(),
            });
        }
        if matches!(op.conflict, ConflictStrategy::Ignore) {
            // Pinecone has no "insert if not exists" primitive.
            return Err(CompileError::OperatorUnsupported {
                backend: BackendKind::Pinecone,
                op: "insert_ignore",
            });
        }
        let table = self.resolve_table(&op.message_type, ctx)?;
        let pk_field = table
            .primary_key
            .first()
            .ok_or_else(|| CompileError::Malformed {
                reason: format!(
                    "Pinecone write on '{}' requires a primary-key field (vector ID)",
                    op.message_type
                ),
            })?;
        // Partial update path: ConflictStrategy::Update { fields } →
        // /vectors/update (single vector, partial metadata + optional
        // values). Update is single-record only on Pinecone's REST
        // surface.
        if let ConflictStrategy::Update { fields, .. } = &op.conflict {
            if op.records.len() != 1 {
                return Err(CompileError::OperatorUnsupported {
                    backend: BackendKind::Pinecone,
                    op: "batch_partial_update",
                });
            }
            let record = &op.records[0];
            let id_value = record
                .get(pk_field)
                .ok_or_else(|| CompileError::Malformed {
                    reason: format!("record missing primary-key field '{pk_field}'"),
                })?;
            let id = id_from_value(id_value)?;
            let mut body = Map::new();
            body.insert("id".into(), json!(id));
            // setMetadata: only the listed fields. Validate each.
            let mut set_md = Map::new();
            for field in fields {
                let resolved = self.field_for(table, field, &op.message_type)?;
                if let Some(v) = record.get(field) {
                    set_md.insert(resolved.to_string(), value_to_pinecone_json(v));
                }
            }
            if !set_md.is_empty() {
                body.insert("setMetadata".into(), Json::Object(set_md));
            }
            // Optional new vector values (when `_vector` is supplied).
            if let Some(LogicalValue::Array(arr)) = record.get("_vector") {
                let values: Vec<f32> = arr
                    .iter()
                    .filter_map(|v| match v {
                        LogicalValue::Float(f) => Some(*f as f32),
                        LogicalValue::Int(i) => Some(*i as f32),
                        _ => None,
                    })
                    .collect();
                body.insert("values".into(), json!(values));
            }
            Self::add_namespace(&mut body, ctx);
            return Ok(CompiledRendering::Json {
                backend: BackendKind::Pinecone,
                method: HttpMethod::Post,
                path: "/vectors/update".to_string(),
                body: Json::Object(body),
            });
        }

        // Standard upsert path: full vector + metadata. Batch
        // supported up to Pinecone's 100-vector / 2 MB request limit
        // (operator-controlled batching upstream).
        let mut vectors: Vec<Json> = Vec::with_capacity(op.records.len());
        for record in &op.records {
            let id_value = record
                .get(pk_field)
                .ok_or_else(|| CompileError::Malformed {
                    reason: format!("record missing primary-key field '{pk_field}'"),
                })?;
            let id = id_from_value(id_value)?;
            let vector = match record.get("_vector") {
                Some(LogicalValue::Array(arr)) => arr
                    .iter()
                    .filter_map(|v| match v {
                        LogicalValue::Float(f) => Some(*f as f32),
                        LogicalValue::Int(i) => Some(*i as f32),
                        _ => None,
                    })
                    .collect::<Vec<f32>>(),
                _ => {
                    return Err(CompileError::Malformed {
                        reason: "Pinecone upsert requires an Array `_vector` field".into(),
                    });
                }
            };
            // Optional sparse values for hybrid-search-ready storage.
            // Convention: `_sparse_indices: [u32]` + `_sparse_values: [f32]`.
            let sparse = match (record.get("_sparse_indices"), record.get("_sparse_values")) {
                (Some(LogicalValue::Array(idx)), Some(LogicalValue::Array(vals))) => {
                    let indices: Vec<i64> = idx
                        .iter()
                        .filter_map(|v| match v {
                            LogicalValue::Int(i) => Some(*i),
                            _ => None,
                        })
                        .collect();
                    let values: Vec<f32> = vals
                        .iter()
                        .filter_map(|v| match v {
                            LogicalValue::Float(f) => Some(*f as f32),
                            LogicalValue::Int(i) => Some(*i as f32),
                            _ => None,
                        })
                        .collect();
                    Some(json!({ "indices": indices, "values": values }))
                }
                _ => None,
            };
            let mut metadata = Map::new();
            for (k, v) in record {
                if k == pk_field
                    || k == "_vector"
                    || k == "_sparse_indices"
                    || k == "_sparse_values"
                {
                    continue;
                }
                metadata.insert(k.clone(), value_to_pinecone_json(v));
            }
            if let Some(tid) = ctx.tenant_id
                && !tid.is_empty()
                && let Some(column) = Some(super::util::tenant_system_field(table))
            {
                metadata.insert(column.to_string(), json!(tid));
            }
            if let Some(pid) = ctx.project_id
                && !pid.is_empty()
                && let Some(column) = Some(super::util::project_system_field(table))
            {
                metadata.insert(column.to_string(), json!(pid));
            }
            let mut vec_obj = json!({
                "id": id,
                "values": vector,
                "metadata": Json::Object(metadata)
            });
            if let Some(s) = sparse {
                vec_obj["sparseValues"] = s;
            }
            vectors.push(vec_obj);
        }
        let mut body = Map::new();
        body.insert("vectors".into(), json!(vectors));
        Self::add_namespace(&mut body, ctx);
        Ok(CompiledRendering::Json {
            backend: BackendKind::Pinecone,
            method: HttpMethod::Post,
            path: "/vectors/upsert".to_string(),
            body: Json::Object(body),
        })
    }

    fn compile_delete(
        &self,
        op: &LogicalDelete,
        ctx: &CompileContext<'_>,
    ) -> Result<CompiledRendering, CompileError> {
        let table = self.resolve_table(&op.message_type, ctx)?;
        let pk_field = table.primary_key.first();
        let mut body = Map::new();
        Self::add_namespace(&mut body, ctx);

        // Branch on filter shape — same three cases as compile_read:
        //   1. PK eq → {ids: [id]}
        //   2. PK IN → {ids: [...]}
        //   3. metadata filter → {filter: {...}}
        if let Some(pk) = pk_field {
            if let Some(pk_value) = Self::extract_pk_value(&op.filter, pk) {
                let id = id_from_value(pk_value)?;
                body.insert("ids".into(), json!([id]));
                return Ok(CompiledRendering::Json {
                    backend: BackendKind::Pinecone,
                    method: HttpMethod::Post,
                    path: "/vectors/delete".to_string(),
                    body: Json::Object(body),
                });
            }
            if let Some(values) = Self::extract_pk_list(&op.filter, pk) {
                let ids: Vec<String> = values
                    .iter()
                    .filter_map(|v| id_from_value(v).ok())
                    .collect();
                body.insert("ids".into(), json!(ids));
                return Ok(CompiledRendering::Json {
                    backend: BackendKind::Pinecone,
                    method: HttpMethod::Post,
                    path: "/vectors/delete".to_string(),
                    body: Json::Object(body),
                });
            }
        }
        // Filter path.
        let user_filter = self.render_filter(&op.filter, table, &op.message_type)?;
        if user_filter.as_object().is_some_and(|m| m.is_empty()) {
            return Err(CompileError::Malformed {
                reason: "LogicalDelete::filter cannot be empty (use Drop resource to truncate \
                         a namespace via deleteAll)"
                    .into(),
            });
        }
        let filter = self.and_with_context(user_filter, table, ctx);
        body.insert("filter".into(), filter);
        Ok(CompiledRendering::Json {
            backend: BackendKind::Pinecone,
            method: HttpMethod::Post,
            path: "/vectors/delete".to_string(),
            body: Json::Object(body),
        })
    }

    fn compile_search(
        &self,
        op: &LogicalSearch,
        ctx: &CompileContext<'_>,
    ) -> Result<CompiledRendering, CompileError> {
        let table = self.resolve_table(&op.message_type, ctx)?;
        let vector = op.vector.as_ref().ok_or_else(|| CompileError::Malformed {
            reason: "Pinecone search requires a dense vector (Pinecone has no native text-only \
                     search; precompute embeddings via the inference API or your own model)"
                .into(),
        })?;
        let user_filter = match &op.filter {
            Some(f) => self.render_filter(f, table, &op.message_type)?,
            None => json!({}),
        };
        let filter = self.and_with_context(user_filter, table, ctx);
        let mut body = Map::new();
        body.insert("vector".into(), json!(vector));
        body.insert("topK".into(), json!(op.top_k));
        body.insert("includeMetadata".into(), json!(op.with_payload));
        body.insert("includeValues".into(), json!(op.with_vector));
        if !filter.as_object().is_some_and(|m| m.is_empty()) {
            body.insert("filter".into(), filter);
        }
        Self::add_namespace(&mut body, ctx);

        // Hybrid sparse+dense search: when text_query is set and the
        // caller supplied sparse values via the filter (convention:
        // a top-level "_sparse_indices"/"_sparse_values" in the
        // filter or attributes), include them as `sparseVector`.
        // Pinecone returns a fused score (Reciprocal Rank Fusion
        // when alpha is set; default is alpha=1.0 → dense-only).
        if let Some(text) = &op.text_query
            && !text.is_empty()
        {
            // Without precomputed sparse values, hybrid search isn't
            // achievable on Pinecone — surface the requirement
            // clearly so callers know they need to supply them.
            // Operators using Pinecone's inference API to generate
            // sparse embeddings call /embed first, then pass the
            // result through. We record the text in the request body
            // as `_text_query` so the executor (or operator wrapper)
            // can drive the inference call.
            body.insert("_text_query".into(), json!(text));
        }

        Ok(CompiledRendering::Json {
            backend: BackendKind::Pinecone,
            method: HttpMethod::Post,
            path: "/query".to_string(),
            body: Json::Object(body),
        })
    }

    fn compile_aggregate(
        &self,
        op: &LogicalAggregate,
        ctx: &CompileContext<'_>,
    ) -> Result<CompiledRendering, CompileError> {
        if op.aggregates.is_empty() {
            return Err(CompileError::Malformed {
                reason: "LogicalAggregate::aggregates must be non-empty".into(),
            });
        }
        super::util::validate_aggregate_aliases(&op.aggregates)?;
        // Pinecone supports ONLY `COUNT(*)` (and group-by namespace)
        // via `/describe_index_stats`. Anything else is a clean
        // refusal so callers know to route elsewhere.
        for agg in &op.aggregates {
            if !matches!(agg.func, AggregateFunc::Count) || agg.field != "*" {
                return Err(CompileError::OperatorUnsupported {
                    backend: BackendKind::Pinecone,
                    op: "non_count_aggregate",
                });
            }
        }
        // GROUP BY support: only the synthetic `namespace` column is
        // permitted (Pinecone's stats are namespace-grouped).
        if !op.group_by.is_empty()
            && !(op.group_by.len() == 1
                && op
                    .group_by
                    .iter()
                    .all(|f| f == "namespace" || f == "_namespace"))
        {
            return Err(CompileError::OperatorUnsupported {
                backend: BackendKind::Pinecone,
                op: "group_by_non_namespace",
            });
        }
        // Optional filter — Pinecone's describe_index_stats accepts a
        // `filter` field that scopes the counts.
        let table = self.resolve_table(&op.message_type, ctx)?;
        let mut body = Map::new();
        if let Some(f) = &op.filter {
            let user = self.render_filter(f, table, &op.message_type)?;
            let filter = self.and_with_context(user, table, ctx);
            if !filter.as_object().is_some_and(|m| m.is_empty()) {
                body.insert("filter".into(), filter);
            }
        } else {
            // Apply tenant filter only.
            let filter = self.and_with_context(json!({}), table, ctx);
            if !filter.as_object().is_some_and(|m| m.is_empty()) {
                body.insert("filter".into(), filter);
            }
        }
        Ok(CompiledRendering::Json {
            backend: BackendKind::Pinecone,
            method: HttpMethod::Post,
            path: "/describe_index_stats".to_string(),
            body: Json::Object(body),
        })
    }

    fn compile_resource_op(
        &self,
        op: &LogicalResourceOp,
        _ctx: &CompileContext<'_>,
    ) -> Result<CompiledRendering, CompileError> {
        let name = &op.resource_name;
        let (method, path, body) = match (op.op, op.resource_kind) {
            // Index lifecycle: create / drop / list.
            (ResourceOpKind::Ensure, ResourceKind::Index)
            | (ResourceOpKind::Ensure, ResourceKind::Collection)
                if matches!(op.resource_kind, ResourceKind::Index) =>
            {
                let mut body = op.spec.clone().unwrap_or_else(|| json!({}));
                if let Json::Object(map) = &mut body {
                    map.entry("name".to_string()).or_insert_with(|| json!(name));
                    map.entry("dimension".to_string())
                        .or_insert_with(|| json!(1536));
                    map.entry("metric".to_string())
                        .or_insert_with(|| json!("cosine"));
                }
                (HttpMethod::Post, "/indexes".to_string(), body)
            }
            (ResourceOpKind::Drop, ResourceKind::Index) => {
                (HttpMethod::Delete, format!("/indexes/{name}"), json!({}))
            }
            (ResourceOpKind::List, ResourceKind::Index) => {
                (HttpMethod::Get, "/indexes".to_string(), json!({}))
            }
            // Collection lifecycle (Pinecone collections = index
            // snapshots; same REST shape under /collections).
            (ResourceOpKind::Ensure, ResourceKind::Collection) => {
                let mut body = op.spec.clone().unwrap_or_else(|| json!({}));
                if let Json::Object(map) = &mut body {
                    map.entry("name".to_string()).or_insert_with(|| json!(name));
                }
                (HttpMethod::Post, "/collections".to_string(), body)
            }
            (ResourceOpKind::Drop, ResourceKind::Collection) => (
                HttpMethod::Delete,
                format!("/collections/{name}"),
                json!({}),
            ),
            (ResourceOpKind::List, ResourceKind::Collection) => {
                (HttpMethod::Get, "/collections".to_string(), json!({}))
            }
            _ => {
                return Err(CompileError::OperatorUnsupported {
                    backend: BackendKind::Pinecone,
                    op: "unsupported_resource_kind",
                });
            }
        };
        Ok(CompiledRendering::Json {
            backend: BackendKind::Pinecone,
            method,
            path,
            body,
        })
    }
}

/// Coerce a LogicalValue to a Pinecone vector ID. Pinecone IDs must
/// be strings, but operators frequently use integer PKs in their
/// schemas; we stringify those.
fn id_from_value(v: &LogicalValue) -> Result<String, CompileError> {
    match v {
        LogicalValue::String(s) => Ok(s.clone()),
        LogicalValue::Int(i) => Ok(i.to_string()),
        _ => Err(CompileError::Malformed {
            reason: "Pinecone vector IDs must be strings or integers".into(),
        }),
    }
}

fn value_to_pinecone_json(v: &LogicalValue) -> Json {
    match v {
        LogicalValue::Null => Json::Null,
        LogicalValue::Bool(b) => Json::Bool(*b),
        LogicalValue::Int(i) => Json::Number((*i).into()),
        LogicalValue::Float(f) => serde_json::Number::from_f64(*f)
            .map(Json::Number)
            .unwrap_or(Json::Null),
        LogicalValue::String(s) => Json::String(s.clone()),
        LogicalValue::Bytes(b) => {
            use base64::Engine as _;
            Json::String(base64::engine::general_purpose::STANDARD.encode(b))
        }
        LogicalValue::Timestamp(t) => Json::String(t.to_rfc3339()),
        LogicalValue::Json(j) => j.clone(),
        LogicalValue::Array(values) => {
            Json::Array(values.iter().map(value_to_pinecone_json).collect())
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::generation::{CatalogManifest, ManifestColumn, ManifestTable};
    use crate::ir::operations::{AggregateExpr, LogicalRecord};
    use crate::ir::value::LogicalValue;

    fn fixture() -> CatalogManifest {
        let table = ManifestTable {
            message_name: "acme.docs.v1.Doc".into(),
            schema: "vectors".into(),
            table: "docs".into(),
            primary_key: vec!["id".into()],
            columns: vec![
                ManifestColumn {
                    field_name: "id".into(),
                    column_name: "id".into(),
                    proto_type: "string".into(),
                    sql_type: "text".into(),
                    is_primary: true,
                    ..Default::default()
                },
                ManifestColumn {
                    field_name: "title".into(),
                    column_name: "title".into(),
                    proto_type: "string".into(),
                    sql_type: "text".into(),
                    ..Default::default()
                },
            ],
            ..Default::default()
        };
        CatalogManifest {
            tables: vec![table],
            ..Default::default()
        }
    }

    fn extract_json(r: CompiledRendering) -> (String, HttpMethod, Json) {
        match r {
            CompiledRendering::Json {
                path, method, body, ..
            } => (path, method, body),
            other => panic!("expected Json, got {other:?}"),
        }
    }

    // ── Existing coverage (regression guards) ─────────────────────

    #[test]
    fn read_pk_uses_vectors_fetch() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let read =
            LogicalRead::message("acme.docs.v1.Doc").with_filter(LogicalFilter::Comparison {
                field: "id".into(),
                op: ComparisonOp::Eq,
                value: LogicalValue::String("doc1".into()),
            });
        let (path, method, _) = extract_json(PineconeCompiler.compile_read(&read, &ctx).unwrap());
        assert_eq!(path, "/vectors/fetch?ids=doc1");
        assert_eq!(method, HttpMethod::Get);
    }

    #[test]
    fn write_emits_upsert_with_metadata_and_tenant() {
        let m = fixture();
        let ctx = CompileContext::new(&m).with_tenant("acme");
        let mut rec = LogicalRecord::new();
        rec.insert("id".into(), LogicalValue::String("v1".into()));
        rec.insert("title".into(), LogicalValue::String("hello".into()));
        rec.insert(
            "_vector".into(),
            LogicalValue::Array(vec![LogicalValue::Float(0.1), LogicalValue::Float(0.2)]),
        );
        let write = LogicalWrite {
            message_type: "acme.docs.v1.Doc".into(),
            records: vec![rec],
            conflict: ConflictStrategy::Replace,
            return_fields: vec![],
        };
        let (path, _, body) = extract_json(PineconeCompiler.compile_write(&write, &ctx).unwrap());
        assert_eq!(path, "/vectors/upsert");
        let v0 = &body["vectors"][0];
        assert_eq!(v0["id"], "v1");
        assert_eq!(v0["values"].as_array().unwrap().len(), 2);
        assert_eq!(v0["metadata"]["title"], "hello");
        assert_eq!(v0["metadata"]["_tenant_id"], "acme");
    }

    #[test]
    fn write_requires_vector_field() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let mut rec = LogicalRecord::new();
        rec.insert("id".into(), LogicalValue::String("v1".into()));
        let write = LogicalWrite {
            message_type: "acme.docs.v1.Doc".into(),
            records: vec![rec],
            conflict: ConflictStrategy::Replace,
            return_fields: vec![],
        };
        let err = PineconeCompiler.compile_write(&write, &ctx).unwrap_err();
        assert!(matches!(err, CompileError::Malformed { .. }));
    }

    // ── NEW: expanded coverage ────────────────────────────────────

    #[test]
    fn read_pk_in_list_batches_fetch_ids() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let read = LogicalRead::message("acme.docs.v1.Doc").with_filter(LogicalFilter::InList {
            field: "id".into(),
            values: vec![
                LogicalValue::String("a".into()),
                LogicalValue::String("b".into()),
                LogicalValue::String("c".into()),
            ],
        });
        let (path, method, _) = extract_json(PineconeCompiler.compile_read(&read, &ctx).unwrap());
        assert_eq!(method, HttpMethod::Get);
        assert_eq!(path, "/vectors/fetch?ids=a,b,c");
    }

    #[test]
    fn read_without_filter_uses_vectors_list_with_namespace() {
        let m = fixture();
        let ctx = CompileContext::new(&m).with_project("billing");
        let read = LogicalRead::message("acme.docs.v1.Doc")
            .with_pagination(crate::ir::projection::LogicalPagination::limit(100));
        let (path, method, body) =
            extract_json(PineconeCompiler.compile_read(&read, &ctx).unwrap());
        assert_eq!(method, HttpMethod::Post);
        assert_eq!(path, "/vectors/list");
        assert_eq!(body["namespace"], "billing");
        assert_eq!(body["limit"], 100);
    }

    #[test]
    fn partial_update_uses_vectors_update_with_set_metadata() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let mut rec = LogicalRecord::new();
        rec.insert("id".into(), LogicalValue::String("v1".into()));
        rec.insert("title".into(), LogicalValue::String("new title".into()));
        let write = LogicalWrite {
            message_type: "acme.docs.v1.Doc".into(),
            records: vec![rec],
            conflict: ConflictStrategy::update(vec!["title".into()]),
            return_fields: vec![],
        };
        let (path, _, body) = extract_json(PineconeCompiler.compile_write(&write, &ctx).unwrap());
        assert_eq!(path, "/vectors/update");
        assert_eq!(body["id"], "v1");
        assert_eq!(body["setMetadata"]["title"], "new title");
        // No `values` since `_vector` wasn't supplied.
        assert!(body.get("values").is_none());
    }

    #[test]
    fn partial_update_with_vector_includes_values() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let mut rec = LogicalRecord::new();
        rec.insert("id".into(), LogicalValue::String("v1".into()));
        rec.insert(
            "_vector".into(),
            LogicalValue::Array(vec![LogicalValue::Float(0.5), LogicalValue::Float(0.6)]),
        );
        let write = LogicalWrite {
            message_type: "acme.docs.v1.Doc".into(),
            records: vec![rec],
            conflict: ConflictStrategy::update(vec![]),
            return_fields: vec![],
        };
        let (_, _, body) = extract_json(PineconeCompiler.compile_write(&write, &ctx).unwrap());
        assert_eq!(body["values"].as_array().unwrap().len(), 2);
    }

    #[test]
    fn write_with_sparse_values_emits_sparse_vector() {
        // Hybrid-ready upsert: caller supplies precomputed sparse
        // embeddings as `_sparse_indices` + `_sparse_values`.
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let mut rec = LogicalRecord::new();
        rec.insert("id".into(), LogicalValue::String("v1".into()));
        rec.insert(
            "_vector".into(),
            LogicalValue::Array(vec![LogicalValue::Float(0.1)]),
        );
        rec.insert(
            "_sparse_indices".into(),
            LogicalValue::Array(vec![LogicalValue::Int(42), LogicalValue::Int(100)]),
        );
        rec.insert(
            "_sparse_values".into(),
            LogicalValue::Array(vec![LogicalValue::Float(0.9), LogicalValue::Float(0.4)]),
        );
        let write = LogicalWrite {
            message_type: "acme.docs.v1.Doc".into(),
            records: vec![rec],
            conflict: ConflictStrategy::Replace,
            return_fields: vec![],
        };
        let (_, _, body) = extract_json(PineconeCompiler.compile_write(&write, &ctx).unwrap());
        let v0 = &body["vectors"][0];
        assert_eq!(v0["sparseValues"]["indices"].as_array().unwrap().len(), 2);
        assert_eq!(v0["sparseValues"]["values"].as_array().unwrap().len(), 2);
    }

    #[test]
    fn search_includes_namespace_when_project_set() {
        let m = fixture();
        let ctx = CompileContext::new(&m).with_project("billing");
        let search = LogicalSearch {
            message_type: "acme.docs.v1.Doc".into(),
            vector: Some(vec![0.1, 0.2]),
            text_query: None,
            filter: None,
            top_k: 5,
            score_threshold: None,
            require_hybrid: false,
            with_vector: false,
            with_payload: true,
        };
        let (_, _, body) = extract_json(PineconeCompiler.compile_search(&search, &ctx).unwrap());
        assert_eq!(body["namespace"], "billing");
    }

    #[test]
    fn search_hybrid_records_text_query() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let search = LogicalSearch {
            message_type: "acme.docs.v1.Doc".into(),
            vector: Some(vec![0.1, 0.2]),
            text_query: Some("rust ownership".into()),
            filter: None,
            top_k: 5,
            score_threshold: None,
            require_hybrid: true,
            with_vector: false,
            with_payload: true,
        };
        let (_, _, body) = extract_json(PineconeCompiler.compile_search(&search, &ctx).unwrap());
        assert_eq!(body["_text_query"], "rust ownership");
    }

    #[test]
    fn delete_by_pk_uses_ids_list() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let del = LogicalDelete {
            message_type: "acme.docs.v1.Doc".into(),
            filter: LogicalFilter::Comparison {
                field: "id".into(),
                op: ComparisonOp::Eq,
                value: LogicalValue::String("doc1".into()),
            },
            return_fields: vec![],
        };
        let (path, _, body) = extract_json(PineconeCompiler.compile_delete(&del, &ctx).unwrap());
        assert_eq!(path, "/vectors/delete");
        assert_eq!(body["ids"].as_array().unwrap()[0], "doc1");
    }

    #[test]
    fn delete_by_pk_in_list_batches_ids() {
        let m = fixture();
        let ctx = CompileContext::new(&m).with_project("billing");
        let del = LogicalDelete {
            message_type: "acme.docs.v1.Doc".into(),
            filter: LogicalFilter::InList {
                field: "id".into(),
                values: vec![
                    LogicalValue::String("a".into()),
                    LogicalValue::String("b".into()),
                ],
            },
            return_fields: vec![],
        };
        let (_, _, body) = extract_json(PineconeCompiler.compile_delete(&del, &ctx).unwrap());
        assert_eq!(body["ids"].as_array().unwrap().len(), 2);
        assert_eq!(body["namespace"], "billing");
    }

    #[test]
    fn delete_empty_filter_rejected() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let del = LogicalDelete {
            message_type: "acme.docs.v1.Doc".into(),
            filter: LogicalFilter::And(vec![]),
            return_fields: vec![],
        };
        let err = PineconeCompiler.compile_delete(&del, &ctx).unwrap_err();
        assert!(matches!(err, CompileError::Malformed { .. }));
    }

    #[test]
    fn aggregate_count_star_uses_describe_index_stats() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let agg = LogicalAggregate {
            message_type: "acme.docs.v1.Doc".into(),
            filter: None,
            group_by: vec![],
            aggregates: vec![AggregateExpr {
                func: AggregateFunc::Count,
                field: "*".into(),
                alias: "total".into(),
            }],
            having: None,
            sort: vec![],
            pagination: None,
        };
        let (path, method, _) =
            extract_json(PineconeCompiler.compile_aggregate(&agg, &ctx).unwrap());
        assert_eq!(path, "/describe_index_stats");
        assert_eq!(method, HttpMethod::Post);
    }

    #[test]
    fn aggregate_group_by_namespace_allowed() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let agg = LogicalAggregate {
            message_type: "acme.docs.v1.Doc".into(),
            filter: None,
            group_by: vec!["namespace".into()],
            aggregates: vec![AggregateExpr {
                func: AggregateFunc::Count,
                field: "*".into(),
                alias: "n".into(),
            }],
            having: None,
            sort: vec![],
            pagination: None,
        };
        assert!(PineconeCompiler.compile_aggregate(&agg, &ctx).is_ok());
    }

    #[test]
    fn aggregate_group_by_other_rejected() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let agg = LogicalAggregate {
            message_type: "acme.docs.v1.Doc".into(),
            filter: None,
            group_by: vec!["title".into()],
            aggregates: vec![AggregateExpr {
                func: AggregateFunc::Count,
                field: "*".into(),
                alias: "n".into(),
            }],
            having: None,
            sort: vec![],
            pagination: None,
        };
        let err = PineconeCompiler.compile_aggregate(&agg, &ctx).unwrap_err();
        assert!(matches!(
            err,
            CompileError::OperatorUnsupported {
                backend: BackendKind::Pinecone,
                op: "group_by_non_namespace"
            }
        ));
    }

    #[test]
    fn aggregate_sum_rejected() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let agg = LogicalAggregate {
            message_type: "acme.docs.v1.Doc".into(),
            filter: None,
            group_by: vec![],
            aggregates: vec![AggregateExpr {
                func: AggregateFunc::Sum,
                field: "title".into(),
                alias: "t".into(),
            }],
            having: None,
            sort: vec![],
            pagination: None,
        };
        let err = PineconeCompiler.compile_aggregate(&agg, &ctx).unwrap_err();
        assert!(matches!(
            err,
            CompileError::OperatorUnsupported {
                backend: BackendKind::Pinecone,
                op: "non_count_aggregate"
            }
        ));
    }

    #[test]
    fn collection_lifecycle_routes_to_collections_endpoint() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let op = LogicalResourceOp {
            op: ResourceOpKind::Ensure,
            resource_kind: ResourceKind::Collection,
            resource_name: "my-snapshot".into(),
            spec: Some(json!({ "source": "my-index" })),
        };
        let (path, method, body) =
            extract_json(PineconeCompiler.compile_resource_op(&op, &ctx).unwrap());
        assert_eq!(path, "/collections");
        assert_eq!(method, HttpMethod::Post);
        assert_eq!(body["name"], "my-snapshot");
        assert_eq!(body["source"], "my-index");
    }

    #[test]
    fn collection_drop_routes_to_named_endpoint() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let op = LogicalResourceOp {
            op: ResourceOpKind::Drop,
            resource_kind: ResourceKind::Collection,
            resource_name: "snap1".into(),
            spec: None,
        };
        let (path, method, _) =
            extract_json(PineconeCompiler.compile_resource_op(&op, &ctx).unwrap());
        assert_eq!(path, "/collections/snap1");
        assert_eq!(method, HttpMethod::Delete);
    }

    #[test]
    fn ensure_index_emits_post_indexes_with_defaults() {
        let m = fixture();
        let ctx = CompileContext::new(&m);
        let op = LogicalResourceOp {
            op: ResourceOpKind::Ensure,
            resource_kind: ResourceKind::Index,
            resource_name: "my-index".into(),
            spec: None,
        };
        let (path, method, body) =
            extract_json(PineconeCompiler.compile_resource_op(&op, &ctx).unwrap());
        assert_eq!(path, "/indexes");
        assert_eq!(method, HttpMethod::Post);
        assert_eq!(body["name"], "my-index");
        assert_eq!(body["dimension"], 1536);
        assert_eq!(body["metric"], "cosine");
    }
}