dynoxide-rs 3.2.0

//! The SearchVectors operation: exact brute-force KNN over a vector index's
//! shadow table.
//!
//! Validation ordering and every pinned error string come from real DynamoDB
//! behaviour captured in eu-west-2 on 2026-08-11, with follow-up captures on
//! 2026-08-12 and 2026-08-13 that were byte-identical in eu-west-2 and
//! us-east-1. Where the captures are silent (noted inline) the behaviour
//! follows the Query family's conventions.
//!
//! Scoring follows the captured precision rule: elements are f32 (the
//! index's storage precision), but all accumulation and the full score
//! computation run in f64, and the final score saturates to the f32 range,
//! so a score that would overflow f32 reports exactly f32::MAX or -f32::MAX,
//! never infinity or NaN (captured from real DynamoDB (eu-west-2 and
//! us-east-1, 2026-08-13)). The zero-magnitude COSINE operand keeps its
//! defined answer (exactly 1.0, captured) for true zero vectors only; an
//! operand whose f32 squares would underflow scores normally under the f64
//! accumulation. Equal scores order by score then
//! base-table primary key; real DynamoDB's tie order is non-deterministic
//! (three identical calls returned three orderings, captured 2026-08-12), so
//! a deterministic local order is a benign documented divergence.

use crate::actions::helpers;
use crate::actions::vector_index::{
    hash_attr, parse_attr_defs, parse_vector_defs, scalar_type_str,
};
use crate::actions::vector_lifecycle::VectorIndexLifecycle;
use crate::errors::{DynoxideError, Result};
use crate::expressions;
use crate::expressions::PathElement;
use crate::expressions::condition::{CompOp, ConditionExpr, Operand};
use crate::storage_backend::{StorageBackend, VectorCandidateRow};
use crate::types::{AttributeValue, Item};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Internal deserialisation struct for detecting missing fields.
///
/// The four required members are declared in the order DynamoDB names them
/// when more than one is missing: IndexName, SearchVector, TableName, TopK
/// (captured eu-west-2 and us-east-1, 23 October 2026). Serde reports the
/// first missing field in declaration order, so the order is load-bearing.
#[derive(Debug, Default, Deserialize)]
struct SearchVectorsRequestRaw {
    /// Required: a missing IndexName, SearchVector, TableName or TopK gets
    /// serde's own "missing field `IndexName` at line 1 column N", which is
    /// DynamoDB's answer too (captured 2026-10-09, six of six regions, and for
    /// TableName eu-west-2 and us-east-1, 9 October 2026), ahead of any
    /// member constraint.
    #[serde(rename = "IndexName")]
    index_name: String,
    /// A bare JSON array of number attribute values, not wrapped in `L`
    /// (confirmed on the wire against the AWS SDK's serialisation).
    #[serde(rename = "SearchVector", deserialize_with = "strict_values::vector")]
    search_vector: Vec<serde_json::Value>,
    #[serde(rename = "TableName")]
    table_name: String,
    /// Deserialised as i32 so a fractional or out-of-i32-range value rejects
    /// inside serde with the raw serde-style SerializationException message
    /// real DynamoDB returns for a fractional TopK: "invalid type: floating
    /// point `3.5`, expected i32 at line 1 column N", captured from real
    /// DynamoDB (eu-west-2 and us-east-1, 2026-08-13). See
    /// `serde_errors::map_serde_to_dynamodb_message` for the pass-through.
    #[serde(rename = "TopK")]
    top_k: i32,
    #[serde(rename = "SearchConditionExpression", default)]
    search_condition_expression: Option<String>,
    #[serde(rename = "ProjectionExpression", default)]
    projection_expression: Option<String>,
    #[serde(rename = "ExpressionAttributeNames", default)]
    expression_attribute_names: Option<HashMap<String, String>>,
    #[serde(
        rename = "ExpressionAttributeValues",
        default,
        deserialize_with = "strict_values::value_map"
    )]
    expression_attribute_values: Option<Vec<(String, serde_json::Value)>>,
    #[serde(rename = "ReturnConsumedCapacity", default)]
    return_consumed_capacity: Option<String>,
}

/// SearchVectors' strict reading of attribute values.
///
/// DynamoDB's SearchVectors deserialiser reads each value as a tagged type,
/// one member at a time, unlike the other operations, which look at a value
/// whole. Everything here was captured from the search endpoint in eu-west-2
/// and us-east-1 on 23 October 2026:
///
/// - A value that is null, has no member or a second one, or is a number, a
///   boolean or a list, at the top or inside a list or map, is a
///   SerializationException with no message.
/// - A member holding the wrong JSON type quotes the deserialiser, position
///   included: "invalid type: integer `1`, expected a string at line 1 column
///   N". B and BS name their type "a Base64-encoded string".
/// - A member DynamoDB does not know, or a bare string where a value belongs,
///   is a ValidationException listing the members it does know, including the
///   six it keeps for itself (see [`INTERNAL_TYPES`]).
///
/// Only the shape is checked here: the value comes back as JSON, for the
/// request decoder to check once the members it reports first are clean.
mod strict_values {
    use serde::Deserialize;
    use serde::de::{self, Deserializer, MapAccess, SeqAccess, Unexpected, Visitor};
    use serde_json::Value;
    use std::collections::HashMap;
    use std::fmt;

    /// The members DynamoDB's deserialiser knows, in the order it lists them.
    const VARIANTS: &[&str] = &[
        "B",
        "BOOL",
        "BS",
        "DECIMAL",
        "DECIMALSET",
        "DOUBLE",
        "DOUBLESET",
        "INT",
        "INTSET",
        "L",
        "M",
        "N",
        "NS",
        "NULL",
        "S",
        "SS",
    ];

    /// The members DynamoDB reads but does not accept in a request, refused as
    /// "This type mapping is not supported" once the value has been read.
    pub(super) const INTERNAL_TYPES: &[&str] = &[
        "DECIMAL",
        "DECIMALSET",
        "DOUBLE",
        "DOUBLESET",
        "INT",
        "INTSET",
    ];

    /// The SerializationException with no message a shapeless value gets.
    fn shapeless<E: de::Error>() -> E {
        E::custom(crate::serde_errors::SERIALIZATION_MARKER)
    }

    /// An attribute value's JSON, checked member by member.
    struct Shape;

    /// A value, wherever one belongs: shaped, kept as JSON.
    struct Element(Value);

    impl<'de> Deserialize<'de> for Element {
        fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
            deserializer.deserialize_any(Shape).map(Element)
        }
    }

    /// B and BS members, whose type DynamoDB names as a Base64-encoded string.
    struct Base64Text(String);

    impl<'de> Deserialize<'de> for Base64Text {
        fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
            struct Text;
            impl Visitor<'_> for Text {
                type Value = String;
                fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
                    f.write_str("a Base64-encoded string")
                }
                fn visit_str<E: de::Error>(self, v: &str) -> Result<String, E> {
                    Ok(v.to_string())
                }
            }
            deserializer.deserialize_string(Text).map(Base64Text)
        }
    }

    impl<'de> Visitor<'de> for Shape {
        type Value = Value;

        fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
            f.write_str("an attribute value")
        }

        fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Value, A::Error> {
            let Some(member) = map.next_key::<String>()? else {
                return Err(shapeless());
            };
            let value = match member.as_str() {
                "S" | "N" | "INT" | "DECIMAL" | "DOUBLE" => Value::String(map.next_value()?),
                "B" => Value::String(map.next_value::<Base64Text>()?.0),
                "BOOL" | "NULL" => Value::Bool(map.next_value()?),
                "SS" | "NS" | "INTSET" | "DECIMALSET" | "DOUBLESET" => {
                    Value::from(map.next_value::<Vec<String>>()?)
                }
                "BS" => Value::from(
                    map.next_value::<Vec<Base64Text>>()?
                        .into_iter()
                        .map(|text| text.0)
                        .collect::<Vec<_>>(),
                ),
                "L" => Value::Array(
                    map.next_value::<Vec<Element>>()?
                        .into_iter()
                        .map(|element| element.0)
                        .collect(),
                ),
                "M" => Value::Object(
                    map.next_value::<HashMap<String, Element>>()?
                        .into_iter()
                        .map(|(name, element)| (name, element.0))
                        .collect(),
                ),
                _ => return Err(de::Error::unknown_variant(&member, VARIANTS)),
            };
            if map.next_key::<de::IgnoredAny>()?.is_some() {
                return Err(shapeless());
            }
            let mut object = serde_json::Map::new();
            object.insert(member, value);
            Ok(Value::Object(object))
        }

        /// A bare string is read as a member's name with no value.
        fn visit_str<E: de::Error>(self, v: &str) -> Result<Value, E> {
            if VARIANTS.contains(&v) {
                Err(E::invalid_type(Unexpected::UnitVariant, &"newtype variant"))
            } else {
                Err(E::unknown_variant(v, VARIANTS))
            }
        }

        fn visit_unit<E: de::Error>(self) -> Result<Value, E> {
            Err(shapeless())
        }
        fn visit_bool<E: de::Error>(self, _: bool) -> Result<Value, E> {
            Err(shapeless())
        }
        fn visit_i64<E: de::Error>(self, _: i64) -> Result<Value, E> {
            Err(shapeless())
        }
        fn visit_u64<E: de::Error>(self, _: u64) -> Result<Value, E> {
            Err(shapeless())
        }
        fn visit_f64<E: de::Error>(self, _: f64) -> Result<Value, E> {
            Err(shapeless())
        }
        fn visit_seq<A: SeqAccess<'de>>(self, _: A) -> Result<Value, A::Error> {
            Err(shapeless())
        }
    }

    /// SearchVector, each element read strictly.
    pub(super) fn vector<'de, D: Deserializer<'de>>(
        deserializer: D,
    ) -> Result<Vec<Value>, D::Error> {
        let values = Vec::<Element>::deserialize(deserializer)?;
        Ok(values.into_iter().map(|value| value.0).collect())
    }

    /// ExpressionAttributeValues, each value read strictly, in the order
    /// written. A null map reads as absent.
    pub(super) fn value_map<'de, D: Deserializer<'de>>(
        deserializer: D,
    ) -> Result<Option<Vec<(String, Value)>>, D::Error> {
        struct Values;
        impl<'de> Visitor<'de> for Values {
            type Value = Vec<(String, Value)>;
            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
                f.write_str("a map")
            }
            fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
                let mut values = Vec::new();
                while let Some((name, value)) = map.next_entry::<String, Element>()? {
                    values.push((name, value.0));
                }
                Ok(values)
            }
        }
        struct Present;
        impl<'de> Visitor<'de> for Present {
            type Value = Option<Vec<(String, Value)>>;
            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
                f.write_str("a map")
            }
            fn visit_none<E: de::Error>(self) -> Result<Self::Value, E> {
                Ok(None)
            }
            fn visit_unit<E: de::Error>(self) -> Result<Self::Value, E> {
                Ok(None)
            }
            fn visit_some<D: Deserializer<'de>>(self, d: D) -> Result<Self::Value, D::Error> {
                d.deserialize_map(Values).map(Some)
            }
        }
        deserializer.deserialize_option(Present)
    }
}

/// Check one expression value DynamoDB has read, as SearchVectors checks it
/// after the members it reports first: a member it keeps for itself, an empty
/// set, then whatever the shared decoder refuses (a number it cannot read, a
/// NULL that is not true). The refusals are bare, with no key named (captured
/// eu-west-2 and us-east-1, 23 October 2026). The error carries the marker
/// prefix the shared decoder uses, for [`crate::serde_errors`] to classify.
fn read_expression_value(raw: serde_json::Value) -> std::result::Result<AttributeValue, String> {
    if let Some(object) = raw.as_object() {
        if object
            .keys()
            .any(|member| strict_values::INTERNAL_TYPES.contains(&member.as_str()))
        {
            return Err(format!(
                "{}This type mapping is not supported",
                crate::serde_errors::VALIDATION_MARKER
            ));
        }
        if let Some(refusal) = object
            .iter()
            .find_map(|(member, value)| crate::types::empty_set_refusal(member, value))
        {
            return Err(format!(
                "{}{refusal}",
                crate::serde_errors::VALIDATION_MARKER
            ));
        }
    }
    AttributeValue::deserialize(raw).map_err(|e| e.to_string())
}

/// Read a search vector element DynamoDB has read: anything but a number is
/// refused when the vector is checked, so only a number's text is kept.
fn read_vector_element(raw: serde_json::Value) -> Option<AttributeValue> {
    match raw {
        serde_json::Value::Object(mut object) if object.len() == 1 => match object.remove("N") {
            Some(serde_json::Value::String(n)) => Some(AttributeValue::N(n)),
            _ => None,
        },
        _ => None,
    }
}

/// A SearchVectors request.
#[derive(Debug, Default)]
pub struct SearchVectorsRequest {
    pub table_name: String,
    pub index_name: String,
    /// The query vector as wire-shaped attribute values; validated against
    /// the index in `execute`.
    pub search_vector: Vec<AttributeValue>,
    pub top_k: i64,
    pub search_condition_expression: Option<String>,
    pub projection_expression: Option<String>,
    pub expression_attribute_names: Option<HashMap<String, String>>,
    pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
    pub return_consumed_capacity: Option<String>,
}

/// The IndexName, SearchVector and TopK member constraints, in the order they
/// are reported.
///
/// The one copy, asked by the request decoder and by `execute`, so a request
/// built in Rust meets the same checks and gets the same words as one decoded
/// from a body.
///
/// The short-name rejection carries no value echo and spells the member path
/// 'IndexName' (captured from real DynamoDB, eu-west-2 and us-east-1,
/// 2026-08-13). The upper bound mirrors the 255 the CreateTable family
/// enforces; its wording for this operation is uncaptured beyond the no-echo
/// form the captured lower bound establishes. An empty vector rejects in the
/// same no-echo, 'SearchVector' member-path form (captured from real
/// DynamoDB, eu-west-2 and us-east-1, 2026-08-13).
///
/// TopK below 1 rejects at the request-model layer, in the newer layer's
/// wording with no value echo (captured 2026-10-09, six of six regions). Only
/// the lower bound lives here: 101 rejects bare at the operation layer with
/// its own captured message, and non-integer values never reach this point
/// (the i32 field rejects them inside serde).
fn member_constraint_errors(index_name: &str, search_vector_len: usize, top_k: i64) -> Vec<String> {
    let mut errors = Vec::new();
    if index_name.len() < 3 {
        errors.push(
            "Value at 'IndexName' failed to satisfy constraint: \
             Member must have length greater than or equal to 3"
                .to_string(),
        );
    } else if index_name.len() > 255 {
        errors.push(
            "Value at 'IndexName' failed to satisfy constraint: \
             Member must have length less than or equal to 255"
                .to_string(),
        );
    }
    if search_vector_len == 0 {
        errors.push(
            "Value at 'SearchVector' failed to satisfy constraint: \
             Member must have length greater than or equal to 1"
                .to_string(),
        );
    }
    if top_k < 1 {
        errors.push(
            "Value at 'TopK' failed to satisfy constraint: \
             Member must have value greater than or equal to 1"
                .to_string(),
        );
    }
    errors
}

impl<'de> serde::Deserialize<'de> for SearchVectorsRequest {
    fn deserialize<D: serde::Deserializer<'de>>(
        deserializer: D,
    ) -> std::result::Result<Self, D::Error> {
        // A member holding the wrong JSON type, an explicit null for a required
        // member included, is refused by DynamoDB's strict deserialiser as a
        // SerializationException quoting it, position included: "invalid type:
        // null, expected a string at line 1 column N" (captured eu-west-2 and
        // us-east-1, 9 and 23 October 2026). Every other operation maps these
        // to its own words, so the pass-through is made here rather than in the
        // shared decoder. A null optional member reads as absent, as it does on
        // DynamoDB.
        // An integer TopK beyond i32 is quoted the same way, as a
        // ValidationException (captured eu-west-2 and us-east-1, 23 October
        // 2026).
        let raw = SearchVectorsRequestRaw::deserialize(deserializer).map_err(|e| {
            let msg = e.to_string();
            let marker = if msg.starts_with("invalid type: ") {
                crate::serde_errors::SERIALIZATION_AS_READ_MARKER
            } else if msg.starts_with("invalid value: ") {
                crate::serde_errors::VALIDATION_AS_READ_MARKER
            } else {
                return e;
            };
            serde::de::Error::custom(format!("{marker}{msg}"))
        })?;
        use crate::validation::format_validation_errors;

        // An empty table name is the newer validation layer's own check,
        // reported on its own (captured 2026-10-09, six of six regions that
        // serve SearchVectors). A missing one never gets this far.
        if raw.table_name.is_empty() {
            return Err(serde::de::Error::custom(format!(
                "VALIDATION:{}",
                crate::validation::envelope_message(crate::validation::EMPTY_TABLE_NAME_REJECTION)
            )));
        }

        let errors = member_constraint_errors(
            &raw.index_name,
            raw.search_vector.len(),
            i64::from(raw.top_k),
        );

        if let Some(msg) = format_validation_errors(&errors) {
            return Err(serde::de::Error::custom(format!("VALIDATION:{msg}")));
        }

        // A table name that breaks its constraints comes after those members
        // and ahead of ReturnConsumedCapacity, without the envelope, one error
        // per name (captured 2026-10-09, six of six regions).
        if let Some(msg) = crate::validation::first_table_name_error(&raw.table_name) {
            return Err(serde::de::Error::custom(format!("VALIDATION:{msg}")));
        }

        if let Some(e) = return_consumed_capacity_error(raw.return_consumed_capacity.as_deref()) {
            return Err(serde::de::Error::custom(format!("VALIDATION:{e}")));
        }

        // Then the TopK ceiling, the search vector and the expression values,
        // as `execute` checks a request built in Rust.
        if let Some(e) = top_k_ceiling_error(i64::from(raw.top_k)) {
            return Err(serde::de::Error::custom(format!("VALIDATION:{e}")));
        }
        let search_vector = raw
            .search_vector
            .into_iter()
            .map(read_vector_element)
            .collect::<Option<Vec<_>>>()
            .ok_or_else(|| {
                serde::de::Error::custom(format!("VALIDATION:{}", invalid_search_vector()))
            })?;
        search_vector_floats(&search_vector)
            .map_err(|e| serde::de::Error::custom(format!("VALIDATION:{e}")))?;
        let expression_attribute_values = raw
            .expression_attribute_values
            .map(|values| {
                values
                    .into_iter()
                    .map(|(name, value)| Ok((name, read_expression_value(value)?)))
                    .collect::<std::result::Result<HashMap<_, _>, String>>()
            })
            .transpose()
            .map_err(serde::de::Error::custom)?;

        Ok(SearchVectorsRequest {
            table_name: raw.table_name,
            index_name: raw.index_name,
            search_vector,
            top_k: i64::from(raw.top_k),
            search_condition_expression: raw.search_condition_expression,
            projection_expression: raw.projection_expression,
            expression_attribute_names: raw.expression_attribute_names,
            expression_attribute_values,
            return_consumed_capacity: raw.return_consumed_capacity,
        })
    }
}

/// A SearchVectors response: ranked results, best first. `SearchResults` is
/// the response's only data key; there is no pagination surface (captured).
///
/// `ConsumedCapacity` carries request bytes alone, with no `CapacityUnits` and
/// no `TableName`, and reads the same under `TOTAL` and `INDEXES`.
#[derive(Debug, Default, Serialize)]
pub struct SearchVectorsResponse {
    #[serde(rename = "SearchResults")]
    pub search_results: Vec<SearchResult>,
    #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
    pub consumed_capacity: Option<crate::types::VectorSearchCapacity>,
}

/// The billable size of a search: the index entries it scanned, held to the
/// 1KB floor.
///
/// Captured against eu-west-2 on 2026-08-18, real DynamoDB bills this on the
/// data the search read rather than on the request. The figure moved with the
/// table's contents and not with the query, and it does not repeat: five
/// identical searches over one unchanged 512-dimension index reported 14214,
/// 13903, 14214, 14214 and 14518. There is no figure to match, so this reports
/// the same quantity deterministically instead.
///
/// The measure is DynamoDB's own item measure over each entry, the same one the
/// write axis is captured against, summed over the entries the search scanned.
/// Each entry's size is computed once at write and stored with the row, so this
/// reads a column rather than rebuilding what it scanned; an earlier cut sized
/// on the serialised length instead, to avoid parsing every entry, and this
/// needs neither compromise.
///
/// The quantity still diverges from AWS by construction, because AWS does not
/// reproduce its own, so it is not evidence in a conformance claim. What has
/// changed is that the unit is now the captured one rather than a stand-in.
///
/// A documented divergence in the same class as the equal-score tie break:
/// deterministic where AWS is not.
fn search_scanned_bytes(candidates: &[VectorCandidateRow]) -> f64 {
    let scanned: i64 = candidates.iter().map(|row| row.entry_bytes).sum();
    crate::types::vector_request_bytes(scanned.max(0) as usize)
}

/// One search hit: the projected item and its score as a JSON double.
#[derive(Debug, Serialize)]
pub struct SearchResult {
    #[serde(rename = "Item")]
    // Number sets in numeric order here, as DynamoDB returns them;
    // pinned by set_order's every_response_item_field test.
    pub item: Item,
    #[serde(rename = "Score")]
    pub score: f64,
}

/// The captured rejection for a query vector whose elements are not all
/// finite 32-bit floats (an L-wrapped element, a non-number, or an
/// out-of-f32-range value all report the same string; captured 2026-08-11
/// and 2026-08-12).
fn invalid_search_vector() -> DynoxideError {
    DynoxideError::ValidationException(
        "Search vector contains invalid values. All values in the search vector must be a \
         32-bit floating-point number attribute"
            .to_string(),
    )
}

/// The captured rejection for a SearchConditionExpression that is not a pure
/// AND of `=` comparisons. Captured for non-equality comparators on both HASH
/// and INLINE_FILTER elements; other disallowed shapes (OR, NOT, BETWEEN, IN,
/// functions, operand pairings without a path and a value) are uncaptured and
/// adopt the same rejection.
fn invalid_comparator() -> DynoxideError {
    DynoxideError::ValidationException(
        "Invalid SearchConditionExpression: Invalid comparator used in SearchConditionExpression"
            .to_string(),
    )
}

/// One `attribute = :value` conjunct of a SearchConditionExpression.
struct EqualityPair {
    /// Resolved top-level attribute name.
    attr: String,
    /// Whether the path descends below the top level. A nested path can never
    /// equal a stored scalar schema value, so such a pair matches nothing
    /// (uncaptured; the top-level name still counts for schema membership).
    nested: bool,
    /// The resolved comparison operand.
    value: AttributeValue,
}

/// Restriction pass over the parsed condition AST: only `=` comparisons
/// joined by AND are legal, each pairing a document path with a value
/// reference. Collects the equality pairs and every referenced top-level
/// attribute name.
fn collect_equalities(
    expr: &ConditionExpr,
    tracker: &expressions::TrackedExpressionAttributes,
    pairs: &mut Vec<EqualityPair>,
    attrs: &mut Vec<String>,
) -> Result<()> {
    match expr {
        ConditionExpr::And(left, right) => {
            collect_equalities(left, tracker, pairs, attrs)?;
            collect_equalities(right, tracker, pairs, attrs)?;
            Ok(())
        }
        ConditionExpr::Comparison { left, op, right } => {
            if *op != CompOp::Eq {
                return Err(invalid_comparator());
            }
            let (path, value_ref) = match (left, right) {
                (Operand::Path(p), Operand::ValueRef(v))
                | (Operand::ValueRef(v), Operand::Path(p)) => (p, v),
                _ => return Err(invalid_comparator()),
            };
            let top = match path.first() {
                Some(PathElement::Attribute(name)) => tracker.resolve_name(name).map_err(|e| {
                    DynoxideError::ValidationException(format!(
                        "Invalid SearchConditionExpression: {e}"
                    ))
                })?,
                // A path starting with an index access cannot parse, so this
                // arm is defensive.
                _ => return Err(invalid_comparator()),
            };
            // An undefined `:value` carries the operation's prefix, matching
            // the `resolve_name` arm above and the captured prefix
            // convention (the prefixed form itself is uncaptured for value
            // references).
            let value = tracker
                .resolve_value(value_ref)
                .map_err(|e| {
                    DynoxideError::ValidationException(format!(
                        "Invalid SearchConditionExpression: {e}"
                    ))
                })?
                .clone();
            attrs.push(top.clone());
            pairs.push(EqualityPair {
                attr: top,
                nested: path.len() > 1,
                value,
            });
            Ok(())
        }
        _ => Err(invalid_comparator()),
    }
}

/// Compute one candidate's score against the query vector. Elements are f32
/// (the index's storage precision), but all accumulation and the full score
/// computation run in f64, and the final score saturates to the f32 range:
/// AWS accumulates in wider-than-f32 precision and clamps the result, so
/// cancelling DOT_PRODUCT terms whose products overflow f32 report exactly 0
/// and an overflowing EUCLIDEAN distance reports exactly f32::MAX rather
/// than infinity (captured from real DynamoDB (eu-west-2 and us-east-1,
/// 2026-08-13)).
fn score_candidate(distance_function: &str, query: &[f32], stored: &[f32]) -> f64 {
    let score = match distance_function {
        "EUCLIDEAN" => {
            let mut sum = 0.0f64;
            for (q, s) in query.iter().zip(stored) {
                let d = f64::from(*q) - f64::from(*s);
                sum += d * d;
            }
            sum.sqrt()
        }
        "DOT_PRODUCT" => {
            // Higher is better; negative values are legal (captured).
            let mut dot = 0.0f64;
            for (q, s) in query.iter().zip(stored) {
                dot += f64::from(*q) * f64::from(*s);
            }
            dot
        }
        // COSINE
        _ => {
            let mut dot = 0.0f64;
            let mut qq = 0.0f64;
            let mut ss = 0.0f64;
            for (q, s) in query.iter().zip(stored) {
                dot += f64::from(*q) * f64::from(*s);
                qq += f64::from(*q) * f64::from(*q);
                ss += f64::from(*s) * f64::from(*s);
            }
            // A zero-magnitude operand on either side scores exactly 1.0
            // (captured 2026-08-12): an explicit branch, never a division.
            // Only a true zero vector lands here: the f64 squares of finite
            // f32 elements cannot underflow to zero, so an operand like
            // [1e-23, 0] scores normally (captured from real DynamoDB
            // (eu-west-2 and us-east-1, 2026-08-13)).
            if qq == 0.0 || ss == 0.0 {
                return 1.0;
            }
            // A self-match must land exactly on the extremal score; rounding
            // through the norms could otherwise leave sign noise near zero.
            if query == stored {
                return 0.0;
            }
            let c = 1.0 - dot / (qq.sqrt() * ss.sqrt());
            // Clamp at the metric boundaries so floating-point noise can
            // never report a negative distance or one beyond opposite.
            c.clamp(0.0, 2.0)
        }
    };
    // Saturate the final score to the f32 range, preserving sign: a value
    // that would overflow becomes exactly f32::MAX or -f32::MAX (captured
    // from real DynamoDB (eu-west-2 and us-east-1, 2026-08-13)).
    score.clamp(-f64::from(f32::MAX), f64::from(f32::MAX))
}

/// Whether this distance function ranks higher scores as better.
fn higher_is_better(distance_function: &str) -> bool {
    distance_function == "DOT_PRODUCT"
}

/// Validate the query vector: element validity first, then the dimension
/// count. The order is captured: an L-wrapped single element reports the
/// invalid-values message even though its element count also mismatches.
/// The query vector as f32s, or the captured refusal for any element that is
/// not a number reading as a finite 32-bit float. Out-of-range values overflow
/// to infinity on conversion; real DynamoDB rejects them with the same
/// invalid-values string (captured 2026-08-12), and rejects every element that
/// is not a number, whatever it is, before the table is looked up (captured
/// eu-west-2 and us-east-1, 23 October 2026).
fn search_vector_floats(search_vector: &[AttributeValue]) -> Result<Vec<f32>> {
    search_vector
        .iter()
        .map(|elem| match elem {
            AttributeValue::N(n) => match n.parse::<f32>() {
                Ok(v) if v.is_finite() => Ok(v),
                _ => Err(invalid_search_vector()),
            },
            _ => Err(invalid_search_vector()),
        })
        .collect()
}

/// ReturnConsumedCapacity outside its enum, checked on its own and reported
/// bare, after the member constraints and the table name and ahead of the TopK
/// ceiling; a member constraint beside it is reported alone (captured
/// eu-west-2 and us-east-1, 23 October 2026).
fn return_consumed_capacity_error(mode: Option<&str>) -> Option<DynoxideError> {
    crate::validation::return_consumed_capacity_rejection(mode)
        .map(DynoxideError::ValidationException)
}

/// TopK beyond the upper bound rejects bare with the value interpolated
/// (captured 2026-08-11: 101), after the member constraints and before the
/// search vector (captured eu-west-2 and us-east-1, 23 October 2026). The
/// lower bound is a member constraint.
fn top_k_ceiling_error(top_k: i64) -> Option<DynoxideError> {
    (top_k > 100).then(|| {
        DynoxideError::ValidationException(format!(
            "Provided TopK value '{top_k}' is out of valid range. The value must be between 1 \
             and 100 inclusive"
        ))
    })
}

/// SearchVectors' checks on its expression parameters, before the
/// expressions are parsed: an empty map, then a malformed key, then a map
/// with no expression to use it, names ahead of values throughout. Names may
/// serve either expression; values only SearchConditionExpression. Every
/// refusal is bare (captured eu-west-2 and us-east-1, 23 October 2026).
fn check_expression_parameters(request: &SearchVectorsRequest) -> Result<()> {
    let refuse = |message: String| Err(DynoxideError::ValidationException(message));
    let names = request.expression_attribute_names.as_ref();
    let values = request.expression_attribute_values.as_ref();
    if names.is_some_and(HashMap::is_empty) {
        return refuse("ExpressionAttributeNames must not be empty".to_string());
    }
    if values.is_some_and(HashMap::is_empty) {
        return refuse("ExpressionAttributeValues must not be empty".to_string());
    }
    for (map, keys, prefix) in [
        (
            "ExpressionAttributeNames",
            names.map(|n| n.keys().collect::<Vec<_>>()),
            '#',
        ),
        (
            "ExpressionAttributeValues",
            values.map(|v| v.keys().collect::<Vec<_>>()),
            ':',
        ),
    ] {
        let mut keys = keys.unwrap_or_default();
        keys.sort();
        if let Some(key) = keys
            .into_iter()
            .find(|key| !key.starts_with(prefix) || !key.is_ascii())
        {
            return refuse(format!(
                "{map} contains invalid key: Syntax error; key: \"{key}\""
            ));
        }
    }
    let condition = request.search_condition_expression.is_some();
    if names.is_some() && !condition && request.projection_expression.is_none() {
        return refuse(
            "ExpressionAttributeNames can only be specified when using expressions".to_string(),
        );
    }
    if values.is_some() && !condition {
        return refuse(
            "ExpressionAttributeValues can only be specified when using expressions".to_string(),
        );
    }
    Ok(())
}

pub async fn execute<S: StorageBackend>(
    storage: &S,
    request: SearchVectorsRequest,
    lifecycle: &VectorIndexLifecycle,
) -> Result<SearchVectorsResponse> {
    // Validate table name format before checking existence, mirroring the
    // Query family.
    // The same table-name answers the request decoder gives: the empty name
    // enveloped, any other name error bare.
    if request.table_name.is_empty() {
        return Err(DynoxideError::ValidationException(
            crate::validation::envelope_message(crate::validation::EMPTY_TABLE_NAME_REJECTION),
        ));
    }
    // The member constraints, as the decoder reports them: ahead of a
    // malformed table name, which is only looked at once the members are
    // clean.
    if let Some(msg) = crate::validation::format_validation_errors(&member_constraint_errors(
        &request.index_name,
        request.search_vector.len(),
        request.top_k,
    )) {
        return Err(DynoxideError::ValidationException(msg));
    }
    if let Some(msg) = crate::validation::first_table_name_error(&request.table_name) {
        return Err(DynoxideError::ValidationException(msg));
    }

    if let Some(e) = return_consumed_capacity_error(request.return_consumed_capacity.as_deref()) {
        return Err(e);
    }
    if let Some(e) = top_k_ceiling_error(request.top_k) {
        return Err(e);
    }

    // Then the search vector, then each expression value, as the request
    // decoder checks them. A value built in Rust goes through the same reading
    // as one decoded from a body, so it meets the same refusals.
    let query_vec = search_vector_floats(&request.search_vector)?;
    for value in request
        .expression_attribute_values
        .iter()
        .flat_map(HashMap::values)
    {
        let raw = serde_json::to_value(value)
            .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
        read_expression_value(raw).map_err(|msg| {
            crate::serde_errors::classify_marked_serde_error(&msg)
                .unwrap_or(DynoxideError::ValidationException(msg))
        })?;
    }

    check_expression_parameters(&request)?;

    // Expression syntax and the equality-only restriction pass run before the
    // table existence check, mirroring the Query family's captured
    // validate-input-first ordering (uncaptured for this operation). Schema
    // membership needs the index definition and runs after resolution below.
    let tracker = expressions::TrackedExpressionAttributes::new(
        &request.expression_attribute_names,
        &request.expression_attribute_values,
    );

    let mut pairs: Vec<EqualityPair> = Vec::new();
    let mut referenced_attrs: Vec<String> = Vec::new();
    let condition = match request.search_condition_expression.as_deref() {
        Some("") => {
            return Err(DynoxideError::ValidationException(
                "Invalid SearchConditionExpression: The expression can not be empty;".to_string(),
            ));
        }
        Some(expr_str) => {
            // The parser applies the shared expression size limit; its errors
            // (and undefined `#name` references) carry the operation's prefix
            // the way FilterExpression errors do on Query. Only the
            // non-equality comparator string is captured with the prefix; the
            // rest of the family adopts it.
            let parsed = expressions::condition::parse(expr_str).map_err(|e| {
                DynoxideError::ValidationException(format!(
                    "Invalid SearchConditionExpression: {e}"
                ))
            })?;
            expressions::condition::validate_name_refs(
                &parsed,
                &request.expression_attribute_names,
            )
            .map_err(|e| {
                DynoxideError::ValidationException(format!(
                    "Invalid SearchConditionExpression: {e}"
                ))
            })?;
            collect_equalities(&parsed, &tracker, &mut pairs, &mut referenced_attrs)?;
            Some(parsed)
        }
        None => None,
    };

    let projection = match request.projection_expression.as_deref() {
        Some("") => {
            return Err(DynoxideError::ValidationException(
                "Invalid ProjectionExpression: The expression can not be empty;".to_string(),
            ));
        }
        Some(proj_expr) => Some(
            expressions::projection::parse(proj_expr)
                .map_err(DynoxideError::ValidationException)?,
        ),
        None => None,
    };

    // A condition on a nested attribute is refused outright, with one message
    // covering both element types (captured eu-west-2, 2026-08-19), before an
    // unused name or value and before the table is looked up, as is a second
    // condition on one attribute (captured eu-west-2 and us-east-1, 23 October
    // 2026). Without
    // this a nested path silently matched nothing, so a malformed search was
    // indistinguishable from one that found nothing.
    if pairs.iter().any(|p| p.nested) {
        return Err(DynoxideError::ValidationException(
            "Invalid SearchConditionExpression: SearchConditionExpression cannot have \
             conditions on nested attributes"
                .to_string(),
        ));
    }

    // Exactly one condition per attribute. Two that contradict, two that agree,
    // and two carrying a type error between them all give this same message,
    // and the order they appear in does not change it (captured eu-west-2,
    // 2026-08-19). Reconciling two scope values would be modelling something
    // the wire never accepts, so the checks below can assume one pair each.
    {
        let mut seen: Vec<&str> = Vec::with_capacity(pairs.len());
        for pair in &pairs {
            if seen.contains(&pair.attr.as_str()) {
                return Err(DynoxideError::ValidationException(
                    "Invalid SearchConditionExpression: SearchConditionExpression must only \
                     contain one condition per attribute"
                        .to_string(),
                ));
            }
            seen.push(&pair.attr);
        }
    }

    // Register what the expressions use, then refuse an unused name or value
    // before the table is looked up, names first (captured eu-west-2 and
    // us-east-1, 23 October 2026).
    if let Some(ref cond) = condition {
        tracker.track_condition_expr(cond);
    }
    if let Some(ref proj) = projection {
        tracker.track_projection_expr(proj);
        expressions::projection::validate(proj, &tracker)
            .map_err(DynoxideError::ValidationException)?;
    }
    tracker.check_unused()?;

    let meta = helpers::require_table_for_item_op(storage, &request.table_name).await?;
    let vixs = parse_vector_defs(&meta)?;
    let Some(vix) = vixs.iter().find(|v| v.index_name == request.index_name) else {
        return Err(DynoxideError::ValidationException(format!(
            "The table does not have the specified index: {}",
            request.index_name
        )));
    };

    // An index added to a live table refuses a search for the whole of its
    // creation window, and keeps refusing for a period after it reports ACTIVE
    // (captured eu-west-2, 2026-08-11). Placed after index resolution so an
    // unknown index still answers its own message, and before the vector and
    // SearchSchema checks: this ordering against them is uncaptured, and is
    // chosen to match the operation's other resource-state answers rather than
    // its input validation.
    if !lifecycle
        .phase_of_index(storage, &request.table_name, &request.index_name)
        .is_searchable()
    {
        return Err(DynoxideError::ValidationException(format!(
            "Cannot search backfilling vector index: {}",
            request.index_name
        )));
    }

    if query_vec.len() != vix.dimensions as usize {
        return Err(DynoxideError::ValidationException(format!(
            "Input search vector dimension {} does not match vector index dimension {}",
            query_vec.len(),
            vix.dimensions
        )));
    }

    // Every referenced attribute must belong to the SearchSchema. The message
    // preserves AWS's own grammar exactly as captured.
    for attr in &referenced_attrs {
        let in_schema = vix
            .search_schema
            .as_ref()
            .is_some_and(|schema| schema.iter().any(|e| e.attribute_name == *attr));
        if !in_schema {
            return Err(DynoxideError::ValidationException(format!(
                "SearchConditionExpression must not contain any attributes that is not in \
                 SearchSchema. Invalid attribute: {attr}"
            )));
        }
    }

    // A HASH-schema index scopes every search to one partition value, so an
    // equality on the HASH attribute is mandatory. Captured for the
    // absent-expression case; an expression that omits the HASH attribute
    // adopts the same string (uncaptured).
    let index_hash_attr = hash_attr(vix);
    if let Some(hash) = index_hash_attr {
        let provided = pairs.iter().any(|p| p.attr == hash && !p.nested);
        if !provided {
            return Err(DynoxideError::ValidationException(
                "SearchConditionExpression must be provided when SearchSchema has a HASH key"
                    .to_string(),
            ));
        }
    }

    // Candidate load, scoped through the indexed hash_value column when the
    // schema declares a HASH element. The operand is encoded with the same
    // key-string encoding the write path stores, so the two sides agree by
    // construction.
    let mut hash_scope: Option<String> = None;
    let mut unscopable_hash = false;
    if let Some(hash) = index_hash_attr {
        let attr_defs = parse_attr_defs(&meta)?;
        // At most one pair survives the duplicate check above, so this settles
        // a single scope value rather than reconciling several.
        for pair in pairs.iter().filter(|p| p.attr == hash) {
            // An operand whose type differs from the attribute's declared
            // AttributeDefinitions type rejects. Captured for an N operand
            // against a declared S (eu-west-2 and us-east-1, 2026-08-13);
            // the format generalises to every other operand type, so the
            // BOOL, L, and M letters are uncaptured within the captured
            // format. CreateTable guarantees a declaration exists for every
            // SearchSchema attribute, so the lookup is defensive only.
            if let Some(def) = attr_defs.iter().find(|d| d.attribute_name == hash) {
                let declared = scalar_type_str(&def.attribute_type);
                let actual = pair.value.type_name();
                if actual != declared {
                    return Err(DynoxideError::ValidationException(format!(
                        "Type of '{hash}' attribute in SearchConditionExpression ({actual}) \
                         does not match type in search schema ({declared})"
                    )));
                }
            }
            // An empty-string operand rejects with a "Key:" suffix, distinct
            // from the write path's IndexName/IndexKey form (captured from
            // real DynamoDB (eu-west-2 and us-east-1, 2026-08-13)). An empty
            // binary operand is uncaptured and falls through: the write path
            // never stores an empty binary hash value, so it matches nothing.
            if matches!(&pair.value, AttributeValue::S(s) if s.is_empty()) {
                return Err(DynoxideError::ValidationException(format!(
                    "One or more parameter values are not valid. The AttributeValue for a key \
                     attribute cannot contain an empty string value. Key: {hash}"
                )));
            }
            // The non-key-able arm of `to_key_string` is unreachable after the
            // type check above, which itself only runs when the attribute has a
            // declaration to check against. It stays as defence in depth, and
            // it fails closed: leaving the scope unset would load every
            // partition rather than none, so a value that cannot be a key
            // matches nothing instead of matching everything.
            match pair.value.to_key_string() {
                Some(key) => hash_scope = Some(key),
                None => unscopable_hash = true,
            }
        }
    }

    let candidates: Vec<VectorCandidateRow> = if unscopable_hash {
        Vec::new()
    } else {
        storage
            .query_vector_candidates(
                &request.table_name,
                &request.index_name,
                hash_scope.as_deref(),
            )
            .await?
    };

    // Score the candidates that pass every INLINE_FILTER equality: the pairs
    // compare wire-shaped values against the stored filter_json entries with
    // the condition machinery's equality (numeric for N). An item without
    // the filter attribute is excluded. HASH pairs were fully settled by the
    // scoped load above.
    let filter_pairs: Vec<&EqualityPair> = pairs
        .iter()
        .filter(|p| Some(p.attr.as_str()) != index_hash_attr)
        .collect();
    let mut scored: Vec<(f64, &VectorCandidateRow)> = Vec::with_capacity(candidates.len());
    for row in &candidates {
        let filter_map: HashMap<String, AttributeValue> = if filter_pairs.is_empty() {
            HashMap::new()
        } else {
            serde_json::from_str(&row.filter_json).map_err(|e| {
                DynoxideError::InternalServerError(format!("Bad filter JSON in storage: {e}"))
            })?
        };
        let passes = filter_pairs.iter().all(|pair| {
            filter_map
                .get(&pair.attr)
                .is_some_and(|stored| expressions::condition::values_equal(stored, &pair.value))
        });
        if !passes {
            continue;
        }

        let stored_vec: Vec<f32> = serde_json::from_str(&row.vector_json).map_err(|e| {
            DynoxideError::InternalServerError(format!("Bad vector JSON in storage: {e}"))
        })?;
        scored.push((
            score_candidate(&vix.distance_function, &query_vec, &stored_vec),
            row,
        ));
    }

    // Rank in the metric's direction with a total-order comparison. The sort
    // is stable and the candidate load is ordered by base-table primary key,
    // so equal scores keep that order: the deterministic tie-break.
    if higher_is_better(&vix.distance_function) {
        scored.sort_by(|a, b| b.0.total_cmp(&a.0));
    } else {
        scored.sort_by(|a, b| a.0.total_cmp(&b.0));
    }
    scored.truncate(request.top_k as usize);

    // Build results from the projected item copies. The vector attribute is
    // excluded by default and returned only when a ProjectionExpression
    // requests it AND the index projects it (the copy is then the index's
    // f32 truncation, captured).
    let loop_tracker = expressions::TrackedExpressionAttributes::without_tracking(
        &request.expression_attribute_names,
        &request.expression_attribute_values,
    );
    let no_keys: &[String] = &[];

    // The projected entries are loaded here, for the survivors alone. The
    // candidate scan leaves them behind, so a search over a large index no
    // longer materialises every entry to score it and drop it.
    let survivor_keys: Vec<(String, String)> = scored
        .iter()
        .map(|(_, row)| (row.table_pk.clone(), row.table_sk.clone()))
        .collect();
    let items: HashMap<(String, String), String> = storage
        .vector_items_for_keys(&request.table_name, &request.index_name, &survivor_keys)
        .await?
        .into_iter()
        .map(|(pk, sk, item_json)| ((pk, sk), item_json))
        .collect();

    let mut search_results = Vec::with_capacity(scored.len());
    for (score, row) in scored {
        let key = (row.table_pk.clone(), row.table_sk.clone());
        // A row that went between the scoring pass and this read is dropped
        // rather than raised. The facade holds the backend for the whole call
        // so it cannot happen there, but a caller driving the storage trait
        // directly can delete underneath a search, and answering a concurrent
        // delete with a 500 would be worse than answering with one result
        // fewer.
        let Some(item_json) = items.get(&key) else {
            continue;
        };
        let item: Item = serde_json::from_str(item_json).map_err(|e| {
            DynoxideError::InternalServerError(format!("Bad item JSON in storage: {e}"))
        })?;
        let result_item = if let Some(ref proj) = projection {
            expressions::projection::apply(&item, proj, &loop_tracker, no_keys)
                .map_err(DynoxideError::ValidationException)?
        } else {
            let mut item = item;
            item.remove(&vix.vector_attribute.attribute_name);
            item
        };
        search_results.push(SearchResult {
            item: result_item,
            score,
        });
    }

    // Identical under TOTAL and INDEXES, absent under NONE (captured).
    let consumed_capacity =
        crate::types::capacity_wanted(request.return_consumed_capacity.as_deref()).then(|| {
            crate::types::VectorSearchCapacity {
                vector_search_request_bytes: search_scanned_bytes(&candidates),
            }
        });

    Ok(SearchVectorsResponse {
        search_results,
        consumed_capacity,
    })
}