yo-resp 0.3.23

//! The search index commands, on the wire (`09` section 5).
//!
//! Sixteen of them are about the index itself rather than about what is in it:
//! making one, changing its schema, taking it away, describing it, listing them
//! and naming them. Two more print the tree a query parses into, and one reads
//! a tag field's values straight back out. Three keep the word lists that are
//! not attached to any index at all. The last is `FT.SEARCH`, which is the only
//! one that reads documents and so the only one that has to reach the keyspace
//! as well as the registry.
//!
//! # There is no key here
//!
//! Every other group in this file's neighbourhood is handed one [`Keyspace`]
//! and works on a key inside it. This one is handed the registry off the server
//! and works on a name that is not a key: an index does not appear in `KEYS *`,
//! `TYPE` has nothing to say about it, and `DEL` cannot remove one. That is
//! Redis's model and not a shortcut, and it is why the registry hangs off
//! [`Server`](super::Server) rather than off a database. `SELECT 1` followed by
//! `FT._LIST` lists the indexes made on database zero, on a real server and on
//! this one.
//!
//! Emptying the keyspace does empty it, though, and from any database:
//! `FLUSHDB` on database nine drops an index that only ever followed keys on
//! database zero. That is measured against a real server rather than reasoned
//! about, and it falls out of the module hanging its callback on the flush
//! event without looking at which database flushed.
//!
//! # The errors are the module's own
//!
//! Nothing here goes through [`Error`](yo_common::Error), which is the type the
//! rest of dispatch answers with and which always comes out under `ERR`,
//! `WRONGTYPE` or `INVALIDOBJ`. The search module has its own code words,
//! `SEARCH_INDEX_EXISTS` and `SEARCH_PARSE_ARGS` and six more, and two of its
//! errors have no code word at all. A client branches on the first word of an
//! error line, so those are copied exactly, down to `Alias does not exist`
//! having no prefix and `Tag separator must be a single character. Got `%s``
//! carrying an unexpanded format specifier that a real server never fills in.
//!
//! That is what [`Fail`] is for. It holds the line in the pieces it is built
//! from, so the one piece that came from the client is the one piece that gets
//! its line endings taken out, and so nothing is allocated to answer a bad
//! argument.
//!
//! # The field option grammar is two loops
//!
//! `SCHEMA t TEXT WITHSUFFIXTRIE SORTABLE` is accepted and
//! `SCHEMA t TEXT SORTABLE WITHSUFFIXTRIE` is not, which is not a typo in
//! either the test or the server. A field is read by two loops, one after the
//! other. The first takes the options that belong to the type, plus
//! `WITHSUFFIXTRIE`, `INDEXEMPTY` and `INDEXMISSING`, in any order. The second
//! takes `SORTABLE [UNF]` and `NOINDEX`, in any order and repeatable, and it
//! never goes back. So a first loop keyword arriving after a second loop
//! keyword is not an option at all: the field is over, and the word is read as
//! the name of the next field, which then has no type after it and says so.
//!
//! This was worked out by measuring a real server rather than by reading its
//! source, and every accept and reject observed follows from it.
//!
//! # What `FT.INFO` cannot say yet
//!
//! Two dozen of its fields are RediSearch's own internals: how many megabytes
//! its inverted index is holding, how many blocks it is in, what its garbage
//! collector has collected. There is no inverted index under this yet and the
//! ones there will be are not theirs, so those fields are reported as zero
//! rather than as an invented number. The four averages are `nan`, which is
//! what a real server answers for an index with no documents in it too, because
//! all four are a division by the document count. D-58 has the whole list.
//!
//! # `FT.SEARCH` runs in two halves
//!
//! Every other command here begins and ends inside the registry lock. This one
//! cannot, because the rows it answers with carry the fields of the keys they
//! came from and reading a key takes a stripe. Holding both locks at once is
//! how a deadlock gets built, and [`indexing`](super::indexing) says so at
//! length, so the command is split. The first half takes the registry, parses
//! the query against the schema, walks it, scores what it found, sorts it and
//! cuts the window the client asked for. What comes out of that is a list of
//! keys and their scores and nothing borrowed from the index. The lock goes,
//! and the second half reads those keys out of the keyspace and writes the
//! reply.
//!
//! A key in the window that will not read is dropped from the reply and taken
//! off the total, which is measured rather than chosen: a document written on
//! database one is indexed and comes back under `NOCONTENT`, and the same
//! search without `NOCONTENT` answers a smaller total with that row missing. So
//! the total a client sees is how many rows the window could actually build,
//! which is why the loading happens before the first byte goes out.

use std::time::{Duration, Instant};

use yo_common::geo;
use yo_common::num::parse_f64;
use yo_common::{Result, parse_i64};
use yo_search::explain::{Note, Why};
use yo_search::field::{self, Algo, Coords, Kind, Tag, Text, Vector, Width};
use yo_search::follow::Errors;
use yo_search::index::{Definition, Sieve, Source};
use yo_search::query::parse::{BAD_POINT, BAD_RADIUS};
use yo_search::query::{self, Ask, Bad, Circle, Mask, Node, Pair, Range, What, Yield};
use yo_search::score::Scorer;
use yo_search::sorted::{self, Sorted};
use yo_search::spell::{self, Lists};
use yo_search::summary::{self, Trim, Wanted, Wrap};
use yo_search::tags::Tags;
use yo_search::walk;
use yo_search::{Clash, English, Field, Index, Registry};
use yo_shape::Metric;

use super::Server;
use super::args::{self, Args};
use super::indexing;
use super::table::Spec;
use crate::reply::Out;

mod aggregate;
mod config;
pub(super) mod cursor;
mod debug;
pub(super) mod docs;
mod hybrid;
mod profile;

use aggregate::{Pipe, Reads, Shape, apply, group, keeps, piped, sorts, windows};
use cursor::{Asks, Kept, Made};
pub(super) use hybrid::hybrid;

/// What somebody profiling a search or an aggregation wants to know about the
/// run, filled in as it goes and read once it is over.
///
/// Nothing here is looked at unless a client asked for it. A search that nobody
/// is watching builds none of it, which is why the walk only puts a tree
/// together when this is present.
#[derive(Default)]
pub(super) struct Watch {
    /// How long the arguments and the query took to read.
    pub(super) parsing: Duration,
    /// How long the query took to turn into something walkable, after it had
    /// been read and before the first document was asked for.
    pub(super) creating: Duration,
    /// How long the walk took, from the first document asked for to the last
    /// one answered.
    pub(super) walking: Duration,
    /// What the walk turned out to be.
    pub(super) ran: Option<walk::Ran>,
    /// Each step the rows went through and how many of them came out of it.
    pub(super) steps: Vec<(Vec<u8>, usize)>,
}

/// The languages a document may be stemmed in, in the spelling `FT.INFO`
/// reports them in.
///
/// The client's spelling is matched against these without regard to case and
/// the entry here is what gets stored, so `LANGUAGE FRENCH` and
/// `LANGUAGE French` both come back as `french`.
const LANGUAGES: &[&[u8]] = &[
    b"arabic",
    b"armenian",
    b"chinese",
    b"danish",
    b"dutch",
    b"english",
    b"finnish",
    b"french",
    b"german",
    b"greek",
    b"hungarian",
    b"indonesian",
    b"irish",
    b"italian",
    b"lithuanian",
    b"nepali",
    b"norwegian",
    b"portuguese",
    b"romanian",
    b"russian",
    b"serbian",
    b"spanish",
    b"swedish",
    b"tamil",
    b"turkish",
    b"yiddish",
];

/// The phonetic matchers a `TEXT` field may ask for.
///
/// Four letters, a two letter algorithm and a two letter language with a colon
/// between them. Only one algorithm exists, so the first half is always `dm`,
/// and the second half is one of these four.
const PHONETICS: &[&[u8]] = &[b"dm:en", b"dm:fr", b"dm:pt", b"dm:es"];

/// One error line, in the pieces it is built from.
///
/// The middle piece is the client's own bytes and the two around it are the
/// server's text, which is the only arrangement any of these lines take. A line
/// with nothing to quote has an empty middle and an empty tail, which writes
/// the head on its own.
///
/// The one exception is a line that was built somewhere else and arrives whole,
/// which is the expression reader's, since where it went wrong and what it was
/// reading are both counted out of the client's own bytes.
struct Fail<'a> {
    head: &'static str,
    word: &'a [u8],
    tail: &'static str,
    whole: Option<Vec<u8>>,
}

impl<'a> Fail<'a> {
    /// A line that says the same thing whatever the client sent.
    const fn plain(head: &'static str) -> Fail<'a> {
        Fail {
            head,
            word: b"",
            tail: "",
            whole: None,
        }
    }

    /// A line that ends with a word the client sent.
    const fn naming(head: &'static str, word: &'a [u8]) -> Fail<'a> {
        Fail {
            head,
            word,
            tail: "",
            whole: None,
        }
    }

    /// A line with a word the client sent in the middle of it.
    const fn about(head: &'static str, word: &'a [u8], tail: &'static str) -> Fail<'a> {
        Fail {
            head,
            word,
            tail,
            whole: None,
        }
    }

    /// A line somebody else wrote, taken as it stands.
    const fn said(whole: Vec<u8>) -> Fail<'a> {
        Fail {
            head: "",
            word: b"",
            tail: "",
            whole: Some(whole),
        }
    }

    /// Writes it out.
    fn write(&self, out: &mut Out) {
        match &self.whole {
            Some(line) => out.error(line),
            None => out.error_about(self.head.as_bytes(), self.word, self.tail.as_bytes()),
        }
    }
}

/// What a command answers when it did not work.
type Answer<'a> = core::result::Result<(), Fail<'a>>;

const EXISTS: &str = "SEARCH_INDEX_EXISTS Index already exists";
/// What `FT.CREATE` says anywhere but on database zero.
///
/// The only sentence in the group with no code word in front of it, which reads
/// like an oversight and is what a real server sends. Only the two creates
/// refuse. `FT.ALTER` and `FT.DROPINDEX` work from any database, and the rest
/// answer the same everywhere, since the registry is one table for the server.
const NOT_DB_ZERO: &str = "Cannot create index on db != 0";
const ALIAS_EXISTS: &str = "SEARCH_INDEX_EXISTS Alias already exists";
const MISSING: &str = "SEARCH_INDEX_NOT_FOUND Index not found: ";
const CONFLICT: &str = "SEARCH_ALIAS_CONFLICT Alias conflicts with an existing index name";
const NOT_MINE: &str = "SEARCH_INDEX_NOT_FOUND Alias does not belong to provided spec";
const NO_ALIAS: &str = "Alias does not exist";
/// What `FT.TAGVALS` says about a name the schema does not have, which is the
/// same line whether the name is nothing at all or the identifier of a field
/// that a query calls something else. It takes the attribute and never the
/// path, so `SCHEMA g AS gg TAG` answers about `gg` and refuses `g`.
const NO_FIELD: &str = "SEARCH_ATTR_BAD No such field";
/// And about a field that is there and is not a tag.
const NOT_A_TAG: &str = "SEARCH_ATTR_BAD Not a tag field";
/// What the two `ALIASADD` spellings answer when the index is not there, which
/// is not the line the rest of the group uses for the same thing.
const NO_TARGET: &str = "SEARCH_INDEX_NOT_FOUND Unknown index name (or name is an alias itself)";
const NO_SCHEMA: &str = "SEARCH_PARSE_ARGS No schema found";
const NO_FIELDS: &str = "SEARCH_PARSE_ARGS Fields arguments are missing";
const AFTER_ALTER: &str = "ALTER must be followed by SCHEMA";
const ALTER_ACTION: &str = "Unknown action passed to ALTER SCHEMA";
const UNKNOWN: &str = "SEARCH_ARG_UNRECOGNIZED Unknown argument `";
const UNKNOWN_BARE: &str = "SEARCH_ARG_UNRECOGNIZED Unknown argument";
const NO_TYPE: &str = "SEARCH_PARSE_ARGS Field `";
const NO_TYPE_END: &str = "` does not have a type";
const BAD_TYPE: &str = "SEARCH_PARSE_ARGS Invalid field type for field `";
const DUPLICATE: &str = "SEARCH_QUERY_BAD Duplicate field in schema - ";
const RULE: &str = "SEARCH_ADD_ARGS Invalid rule type: ";
const LANGUAGE: &str = "SEARCH_ADD_ARGS Invalid language";
const SCORE: &str = "SEARCH_ADD_ARGS Invalid score";
const BOTH: &str =
    "SEARCH_PARSE_ARGS 'Field cannot be defined with both `NOINDEX` and `INDEXMISSING` `";
const SEPARATOR: &str = "SEARCH_PARSE_ARGS Tag separator must be a single character. Got `%s`";
const MATCHER: &str = "SEARCH_QUERY_BAD Matcher Format: <2 chars algorithm>:<2 chars language>. Support algorithms: double metaphone (dm). Supported languages: English (en), French (fr), Portuguese (pt) and Spanish (es)";
const AS_ARG: &str = "SEARCH_PARSE_ARGS AS requires an argument";
const TOO_MANY_TEXT: &str = "SEARCH_QUERY_BAD MAXTEXTFIELDS cannot be used with NOFIELDS";
const BAD_ARGS: &str = "SEARCH_PARSE_ARGS Bad arguments for ";
const NOT_ENOUGH: &str =
    "SEARCH_PARSE_ARGS Bad arguments for vector similarity: not enough arguments";
/// The two halves of the vector grammar that name no keyword of their own, which
/// a real server writes as though they were keywords.
const ALGO_WORD: &str = "vector similarity algorithm";
const COUNT_WORD: &str = "vector similarity number of parameters";
/// What `TRAINING_THRESHOLD` and `REDUCE` answer when they were sent without the
/// compression they only mean something with.
const NO_TRAINING: &str =
    "SEARCH_PARSE_ARGS TRAINING_THRESHOLD is irrelevant when compression was not requested";
const NO_REDUCE: &str =
    "SEARCH_PARSE_ARGS REDUCE is irrelevant when compression is not of type LeanVec";
const SMALL_TRAINING: &str =
    "SEARCH_PARSE_ARGS Invalid TRAINING_THRESHOLD: cannot be lower than DEFAULT_BLOCK_SIZE (1024)";

/// The arity line the two `DROPINDEX` spellings answer with, which names the
/// command with an underscore in front of it. That is not a typo here: a real
/// server answers `'_FT.DROPINDEX'` and `'_FT._DROPINDEXIFX'`, because the
/// module registers the pair under those names and the coordinator wraps them.
const ARITY: &str = "ERR wrong number of arguments for '_";
/// The same line without the underscore, which is what everything else uses.
const ARITY_PLAIN: &str = "ERR wrong number of arguments for '";
/// The tail of both.
const ARITY_END: &str = "' command";

/// The three ways a value can be wrong, in the words a real server ends the
/// `Bad arguments for X` lines with.
const NOT_A_NUMBER: &str = ": Could not convert argument to expected type";
const OUT_OF_RANGE: &str = ": Value is outside acceptable bounds";
const NOT_THERE: &str = ": Expected an argument, but none provided";
const UNKNOWN_WORD: &str = ": Unknown argument";

/// The same four, for the vector lines that quote the keyword in backticks. The
/// closing one belongs to the tail because a line is a head, a client word and a
/// tail, and the keyword is the client's word.
const V_NOT_A_NUMBER: &str = "`: Could not convert argument to expected type";
const V_OUT_OF_RANGE: &str = "`: Value is outside acceptable bounds";
const V_NOT_THERE: &str = "`: Expected an argument, but none provided";
const V_UNKNOWN: &str = "`: Unknown argument";

/// `Bad arguments for X: why`, where `X` is the keyword the value belonged to.
fn bad(what: &'static str, why: &'static str) -> Fail<'static> {
    Fail::about(BAD_ARGS, what.as_bytes(), why)
}

/// An index whose keys have to be read, and whether it may refuse.
///
/// Two commands ask for a scan and they do not ask for the same thing.
/// `FT.CREATE` obeys the `SKIPINITIALSCAN` on the index, since that is what the
/// word means there. `FT.SYNUPDATE` does not: it has a `SKIPINITIALSCAN` of its
/// own, and when the client leaves it off the index is read again whatever it
/// was created with. That is measured, and it is why this carries the answer
/// rather than the registry working it out from the definition.
pub(super) struct Fill<'a> {
    /// The index, by the exact name it was made under.
    pub(super) name: &'a [u8],
    /// Whether the index's own flag has a say in it.
    pub(super) obeys: bool,
}

/// What the caller has to do to the keyspace once the registry lock is gone.
///
/// Both of these want a stripe and neither can have one here, because the whole
/// group runs with the registry held and taking a stripe under it is the lock
/// order a write does not use. So the work comes back out and is done above.
pub(super) enum After<'a> {
    /// An index that was just made and wants the keys already there read in.
    Scan(Fill<'a>),
    /// An index that was just dropped and wants the keys it followed deleted.
    Sweep(Vec<Box<[u8]>>),
}

pub(super) fn execute<'a>(
    server: &Server,
    reg: &mut Registry,
    db: usize,
    spec: &Spec,
    args: Args<'a>,
    out: &mut Out,
) -> Result<Option<After<'a>>> {
    // The one command in the group that is a container of subcommands, so it
    // writes its own arity and unknown subcommand lines rather than answering
    // in the `Fail` shape the rest of these share.
    if spec.name == "FT.CONFIG" {
        config::run(reg, args, out)?;
        return Ok(None);
    }
    // The other one, which writes six error lines of its own on top of those
    // two and answers about the structures rather than about the definition.
    if spec.name == "_FT.DEBUG" {
        debug::run(reg, args, out)?;
        return Ok(None);
    }
    // The keyspace work the caller has to do, which stays `None` for the twelve
    // commands that ask for nothing of the sort, for a create that answered that
    // the name is taken and for a drop that was told to keep the documents.
    let mut made = None;
    let done = match spec.name {
        "FT.CREATE" => create(reg, db, args, out, false).map(|next| made = next),
        "FT._CREATEIFNX" => create(reg, db, args, out, true).map(|next| made = next),
        "FT.ALTER" => alter(reg, args, out, false),
        "FT._ALTERIFNX" => alter(reg, args, out, true),
        "FT.DROPINDEX" => drop_index(reg, spec, args, out, false, true).map(|next| made = next),
        "FT._DROPINDEXIFX" => drop_index(reg, spec, args, out, true, true).map(|next| made = next),
        "FT.DROP" => drop_index(reg, spec, args, out, false, false).map(|next| made = next),
        "FT._DROPIFX" => drop_index(reg, spec, args, out, true, false).map(|next| made = next),
        "FT.INFO" => info(server, reg, args, out),
        "FT._LIST" => list(reg, spec, args, out),
        "FT.ALIASADD" => alias_add(reg, args, out, false),
        "FT._ALIASADDIFNX" => alias_add(reg, args, out, true),
        "FT.ALIASDEL" => alias_del(reg, args, out, false),
        "FT._ALIASDELIFX" => alias_del(reg, args, out, true),
        "FT.ALIASUPDATE" => alias_update(reg, args, out),
        "FT.ALIASLIST" => alias_list(reg, args, out),
        "FT.EXPLAIN" => explain(reg, args, out, false),
        "FT.EXPLAINCLI" => explain(reg, args, out, true),
        "FT.TAGVALS" => tag_vals(reg, args, out),
        "FT.DICTADD" => dict_add(reg, args, out),
        "FT.DICTDEL" => dict_del(reg, args, out),
        "FT.DICTDUMP" => dict_dump(reg, args, out),
        "FT.SYNUPDATE" => syn_update(reg, args, out).map(|next| made = next),
        "FT.SYNDUMP" => syn_dump(reg, args, out),
        "FT.SPELLCHECK" => spellcheck(reg, args, out),
        other => unreachable!("{other} is not a search command"),
    };
    if let Err(f) = done {
        f.write(out);
        return Ok(None);
    }
    // The cursors an index was holding go with it, whichever of the four
    // spellings of the drop took it away.
    if matches!(
        spec.name,
        "FT.DROPINDEX" | "FT._DROPINDEXIFX" | "FT.DROP" | "FT._DROPIFX"
    ) {
        server.cursors.lock().strays(reg);
    }
    Ok(made)
}

/// `FT.CREATE index [options] SCHEMA field type [options] ...`
///
/// The name is checked before the arguments are, which is a real server's order
/// and is visible: `FT.CREATE i BOGUS SCHEMA t TEXT` over an index that already
/// exists answers that it exists rather than that `BOGUS` is not an argument.
///
/// The name comes back when an index was made, because the keys that already
/// match its prefix have to be read into it and this is the only place that
/// knows one was made. `FT._CREATEIFNX` over a name that is taken answers `OK`
/// and hands back nothing, since there is nothing new to fill.
fn create<'a>(
    reg: &mut Registry,
    db: usize,
    args: Args<'a>,
    out: &mut Out,
    ifnx: bool,
) -> core::result::Result<Option<After<'a>>, Fail<'a>> {
    let name = args.get(1);
    // The `IFNX` shortcut comes first and the database comes second, which is
    // the order a real server checks them in and is visible: `FT._CREATEIFNX`
    // over a name that is taken answers `OK` from database one, while
    // `FT.CREATE` over the same name from database one refuses on the database
    // rather than on the name.
    if ifnx && reg.named(name).is_some() {
        out.ok();
        return Ok(None);
    }
    if db != 0 {
        return Err(Fail::plain(NOT_DB_ZERO));
    }
    if reg.named(name).is_some() {
        return Err(Fail::plain(EXISTS));
    }

    let (definition, at) = definition(args, 2)?;
    let mut schema = Vec::new();
    fields(args, at, &mut schema)?;

    // The name is free and the arguments parsed, so nothing below can fail and
    // leave half an index behind.
    let _ = reg.create(Index::new(name, definition, schema));
    out.ok();
    Ok(Some(After::Scan(Fill { name, obeys: true })))
}

/// The options in front of `SCHEMA`, and where the schema starts.
///
/// Any order, which is what a real server takes. Every keyword that carries a
/// count takes exactly that many words after it whatever they say, so
/// `PREFIX 2 a: SCHEMA t TEXT` reads `SCHEMA` as the second prefix and then
/// trips over `t`, which is the error a real server answers too.
fn definition(args: Args<'_>, from: usize) -> core::result::Result<(Definition, usize), Fail<'_>> {
    let mut d = Definition::default();
    let mut prefixes: Option<Vec<Box<[u8]>>> = None;
    let mut at = from;
    loop {
        let Some(a) = args.opt(at) else {
            return Err(Fail::plain(NO_SCHEMA));
        };
        at += 1;
        if args::is(a, b"schema") {
            break;
        } else if args::is(a, b"on") {
            let v = value(args, &mut at, "ON")?;
            d.on = if args::is(v, b"hash") {
                Source::Hash
            } else if args::is(v, b"json") {
                Source::Json
            } else {
                return Err(Fail::naming(RULE, v));
            };
        } else if args::is(a, b"prefix") {
            let n = count(args, &mut at, "PREFIX")?;
            let mut list = Vec::with_capacity(n);
            for _ in 0..n {
                list.push(value(args, &mut at, "PREFIX")?.into());
            }
            prefixes = Some(list);
        } else if args::is(a, b"filter") {
            let src = value(args, &mut at, "FILTER")?;
            // Read here and thrown away, only so that a filter that will not
            // read refuses the create. It is refused where it is written rather
            // than at the end, which is measured: a create with a bad filter and
            // no `SCHEMA` at all answers about the filter.
            Sieve::parse(src).map_err(Fail::said)?;
            d.filter = Some(src.into());
        } else if args::is(a, b"language") {
            let v = value(args, &mut at, "LANGUAGE")?;
            d.language = Some(language(v).ok_or(Fail::plain(LANGUAGE))?.into());
        } else if args::is(a, b"language_field") {
            d.language_field = Some(value(args, &mut at, "LANGUAGE_FIELD")?.into());
        } else if args::is(a, b"score") {
            let v = value(args, &mut at, "SCORE")?;
            // Bounded to a fraction, both ends included, and anything else is
            // the same error as text that is not a number at all.
            d.score = match parse_f64(v) {
                Some(s) if (0.0..=1.0).contains(&s) => s,
                _ => return Err(Fail::plain(SCORE)),
            };
        } else if args::is(a, b"score_field") {
            d.score_field = Some(value(args, &mut at, "SCORE_FIELD")?.into());
        } else if args::is(a, b"payload_field") {
            d.payload_field = Some(value(args, &mut at, "PAYLOAD_FIELD")?.into());
        } else if args::is(a, b"temporary") {
            d.temporary = Some(count(args, &mut at, "TEMPORARY")? as u64);
        } else if args::is(a, b"stopwords") {
            let n = count(args, &mut at, "STOPWORDS")?;
            let mut list = Vec::with_capacity(n);
            for _ in 0..n {
                list.push(value(args, &mut at, "STOPWORDS")?.into());
            }
            d.stopwords = Some(list);
        } else if args::is(a, b"maxtextfields") {
            d.options.maxtextfields = true;
        } else if args::is(a, b"nooffsets") {
            d.options.nooffsets = true;
        } else if args::is(a, b"nohl") {
            d.options.nohl = true;
        } else if args::is(a, b"nofields") {
            d.options.nofields = true;
        } else if args::is(a, b"nofreqs") {
            d.options.nofreqs = true;
        } else if args::is(a, b"skipinitialscan") {
            d.skip_initial_scan = true;
        } else {
            return Err(Fail::about(UNKNOWN, a, "`"));
        }
    }
    // Room for more text fields than the sixteen a header has bits for, next to
    // an order to keep no field bits at all, is a pair that cannot both be
    // honoured. Either order is refused and the sentence names the first one.
    if d.options.maxtextfields && d.options.nofields {
        return Err(Fail::plain(TOO_MANY_TEXT));
    }
    // An index that named no prefixes and one that named none out of a count of
    // zero are the same index, and both follow every key. The one empty prefix
    // is what `FT.INFO` reports for both.
    if let Some(list) = prefixes
        && !list.is_empty()
    {
        d.prefixes = list;
    }
    Ok((d, at))
}

/// The language this spelling means, in the spelling that gets stored.
fn language(v: &[u8]) -> Option<&'static [u8]> {
    LANGUAGES.iter().copied().find(|l| args::is(v, l))
}

/// The next word, or the line that says the keyword ran out of arguments.
fn value<'a>(
    args: Args<'a>,
    at: &mut usize,
    what: &'static str,
) -> core::result::Result<&'a [u8], Fail<'a>> {
    let v = args.opt(*at).ok_or_else(|| bad(what, NOT_THERE))?;
    *at += 1;
    Ok(v)
}

/// The next word as a count of the words after it.
fn count<'a>(
    args: Args<'a>,
    at: &mut usize,
    what: &'static str,
) -> core::result::Result<usize, Fail<'a>> {
    let v = value(args, at, what)?;
    let n = parse_i64(v).ok_or_else(|| bad(what, NOT_A_NUMBER))?;
    usize::try_from(n).map_err(|_| bad(what, OUT_OF_RANGE))
}

/// Every field after `SCHEMA`, appended to what is already there.
///
/// `into` arrives holding the fields the index already has, which is empty for
/// `FT.CREATE` and the whole current schema for `FT.ALTER`, so a field that
/// clashes with an old one and a field that clashes with a new one are found
/// the same way and in the order the client wrote them.
fn fields<'a>(args: Args<'a>, from: usize, into: &mut Vec<Field>) -> Answer<'a> {
    if args.opt(from).is_none() {
        return Err(Fail::plain(NO_FIELDS));
    }
    let mut at = from;
    while at < args.len() {
        let f = one_field(args, &mut at, into)?;
        into.push(f);
    }
    Ok(())
}

/// One field: where it reads from, what a query calls it, what it holds and
/// what may be asked of it.
fn one_field<'a>(
    args: Args<'a>,
    at: &mut usize,
    have: &[Field],
) -> core::result::Result<Field, Fail<'a>> {
    let identifier = args.get(*at);
    *at += 1;
    let attribute = if args.opt(*at).is_some_and(|a| args::is(a, b"as")) {
        *at += 1;
        let named = args.opt(*at).ok_or(Fail::plain(AS_ARG))?;
        *at += 1;
        named
    } else {
        identifier
    };

    let kind = kind(args, at, attribute)?;
    let mut f = Field::new(identifier, kind).named(attribute);
    if have.iter().any(|o| o.attribute == f.attribute) {
        return Err(Fail::naming(DUPLICATE, attribute));
    }

    // The first loop: everything that belongs to the type, plus the three that
    // belong to any of them. It stops at the first word it does not know,
    // which the second loop then gets a look at.
    while let Some(a) = args.opt(*at) {
        if args::is(a, b"withsuffixtrie") && f.kind.takes_empty() {
            f.suffix_trie = true;
        } else if args::is(a, b"indexempty") && f.kind.takes_empty() {
            f.index_empty = true;
        } else if args::is(a, b"indexmissing") {
            f.index_missing = true;
        } else if !type_option(args, at, &mut f)? {
            break;
        }
        // One step past whatever was taken. A bare word leaves the cursor on
        // itself and a pair leaves it on its value, so the same step works for
        // both and the loop cannot stand still.
        *at += 1;
    }

    // The second loop, which never hands anything back to the first one. A
    // `WITHSUFFIXTRIE` down here is not an option, it is the name of the next
    // field.
    while let Some(a) = args.opt(*at) {
        if args::is(a, b"sortable") {
            f.sortable = true;
            *at += 1;
            if args.opt(*at).is_some_and(|n| args::is(n, b"unf")) {
                f.unf = true;
                *at += 1;
            }
        } else if args::is(a, b"noindex") {
            f.noindex = true;
            *at += 1;
        } else {
            break;
        }
    }

    // A field nobody indexes cannot record which documents are missing it,
    // because recording that is indexing it.
    if f.noindex && f.index_missing {
        return Err(Fail::about(BOTH, attribute, "` '"));
    }
    Ok(f)
}

/// What the field holds, read off the word after its name.
fn kind<'a>(
    args: Args<'a>,
    at: &mut usize,
    attribute: &'a [u8],
) -> core::result::Result<Kind, Fail<'a>> {
    let Some(t) = args.opt(*at) else {
        return Err(Fail::about(NO_TYPE, attribute, NO_TYPE_END));
    };
    *at += 1;
    if args::is(t, b"text") {
        Ok(Kind::Text(Text::default()))
    } else if args::is(t, b"tag") {
        Ok(Kind::Tag(Tag::default()))
    } else if args::is(t, b"numeric") {
        Ok(Kind::Numeric)
    } else if args::is(t, b"geo") {
        Ok(Kind::Geo)
    } else if args::is(t, b"geoshape") {
        // The one type whose own option is a bare word rather than a pair, and
        // the only one where leaving it out picks the more expensive default.
        let coords = match args.opt(*at) {
            Some(c) if args::is(c, b"flat") => {
                *at += 1;
                Coords::Flat
            }
            Some(c) if args::is(c, b"spherical") => {
                *at += 1;
                Coords::Spherical
            }
            _ => Coords::Spherical,
        };
        Ok(Kind::GeoShape(coords))
    } else if args::is(t, b"vector") {
        vector(args, at).map(Kind::Vector)
    } else {
        Err(Fail::about(BAD_TYPE, attribute, "`"))
    }
}

/// One option that only means something for one kind of field.
///
/// Answers whether it took the word, so the loop above can tell an option it
/// does not know from an option it does. The cursor is left on the keyword and
/// moved by the caller, which is what keeps the two kinds of option in one
/// loop.
fn type_option<'a>(
    args: Args<'a>,
    at: &mut usize,
    f: &mut Field,
) -> core::result::Result<bool, Fail<'a>> {
    let a = args.get(*at);
    match &mut f.kind {
        Kind::Text(t) => {
            if args::is(a, b"nostem") {
                t.nostem = true;
            } else if args::is(a, b"weight") {
                let mut next = *at + 1;
                let v = value(args, &mut next, "weight")?;
                t.weight = parse_f64(v).ok_or_else(|| bad("weight", NOT_A_NUMBER))?;
                *at = next - 1;
            } else if args::is(a, b"phonetic") {
                let mut next = *at + 1;
                let v = value(args, &mut next, "PHONETIC")?;
                if !PHONETICS.iter().any(|p| args::is(v, p)) {
                    return Err(Fail::plain(MATCHER));
                }
                t.phonetic = Some(v.into());
                *at = next - 1;
            } else {
                return Ok(false);
            }
            Ok(true)
        }
        Kind::Tag(t) => {
            if args::is(a, b"casesensitive") {
                t.casesensitive = true;
            } else if args::is(a, b"separator") {
                let mut next = *at + 1;
                let v = value(args, &mut next, "SEPARATOR")?;
                // One character and not one byte, which for a separator is the
                // same thing: a real server takes the first byte and refuses
                // anything longer, so a multi byte character is refused too.
                let [c] = v else {
                    return Err(Fail::naming(SEPARATOR, v));
                };
                t.separator = *c;
                *at = next - 1;
            } else {
                return Ok(false);
            }
            Ok(true)
        }
        _ => Ok(false),
    }
}

/// `VECTOR algorithm count key value ...`
///
/// The count is a count of words rather than of pairs, and it is the only thing
/// that decides where the vector options end. Words past it belong to the next
/// field and words the schema meant for the next field are read as options when
/// the count reaches over them, which is why an eight over six real options
/// answers about the field name that followed rather than about the count.
///
/// An odd count is not an error of its own. The last keyword it covers is simply
/// left without a value, and that is the line a real server answers with.
fn vector<'a>(args: Args<'a>, at: &mut usize) -> core::result::Result<Vector, Fail<'a>> {
    let a = args.opt(*at).ok_or_else(|| bad(ALGO_WORD, NOT_THERE))?;
    *at += 1;
    let (algo, label) = if args::is(a, b"flat") {
        (Algo::Flat, "FLAT")
    } else if args::is(a, b"hnsw") {
        (Algo::Hnsw, "HNSW")
    } else if args::is(a, b"svs-vamana") {
        (Algo::Svs, "SVS-VAMANA")
    } else {
        return Err(bad(ALGO_WORD, UNKNOWN_WORD));
    };

    let c = args.opt(*at).ok_or_else(|| bad(COUNT_WORD, NOT_THERE))?;
    *at += 1;
    let n = parse_i64(c).ok_or_else(|| bad(COUNT_WORD, NOT_A_NUMBER))?;
    let words = usize::try_from(n).map_err(|_| bad(COUNT_WORD, OUT_OF_RANGE))?;
    if args.len() - *at < words {
        return Err(Fail::plain(NOT_ENOUGH));
    }

    let mut width = None;
    let mut dim = None;
    let mut metric = None;
    let mut training: Option<u64> = None;
    let mut reduce = false;
    let mut v = Vector::new(algo, Width::Float32, 0, Metric::L2);
    let mut left = words;
    while left > 0 {
        let key = args.get(*at);
        *at += 1;
        left -= 1;
        if left == 0 {
            return Err(vbad(label, key, V_NOT_THERE));
        }
        let val = args.get(*at);
        *at += 1;
        left -= 1;

        // The three every algorithm needs, then the ones that belong to one of
        // them. A keyword another algorithm would have taken is not an unknown
        // word, it is a word in the wrong place, and it gets its own line.
        if args::is(key, b"type") {
            width = Some(self::width(val).ok_or_else(|| vbad(label, key, V_UNKNOWN))?);
        } else if args::is(key, b"dim") {
            let d = parse_i64(val).ok_or_else(|| vbad(label, key, V_NOT_A_NUMBER))?;
            if d <= 0 {
                return Err(vbad(label, key, V_OUT_OF_RANGE));
            }
            dim = Some(d as u64);
        } else if args::is(key, b"distance_metric") {
            metric = Some(self::metric(val).ok_or_else(|| vbad(label, key, V_UNKNOWN))?);
        } else if args::is(key, b"initial_cap") && algo != Algo::Svs {
            v.initial_cap = Some(whole(label, key, val)?);
        } else if args::is(key, b"block_size") && algo == Algo::Flat {
            v.block_size = Some(whole(label, key, val)?);
        } else if args::is(key, b"m") && algo == Algo::Hnsw {
            v.m = whole(label, key, val)?;
        } else if args::is(key, b"ef_construction") && algo == Algo::Hnsw {
            v.ef_construction = whole(label, key, val)?;
        } else if args::is(key, b"ef_runtime") && algo == Algo::Hnsw {
            v.ef_runtime = whole(label, key, val)?;
        } else if args::is(key, b"epsilon") && algo != Algo::Flat {
            v.epsilon = parse_f64(val).ok_or_else(|| vbad(label, key, V_NOT_A_NUMBER))?;
        } else if args::is(key, b"graph_max_degree") && algo == Algo::Svs {
            v.graph_max_degree = whole(label, key, val)?;
        } else if args::is(key, b"construction_window_size") && algo == Algo::Svs {
            v.construction_window = whole(label, key, val)?;
        } else if args::is(key, b"search_window_size") && algo == Algo::Svs {
            // Taken, checked and dropped, which is what a real server does with
            // it: nothing it changes reaches `FT.INFO`.
            let _ = whole(label, key, val)?;
        } else if args::is(key, b"compression") && algo == Algo::Svs {
            let c = compression(val).ok_or_else(|| vbad(label, key, V_UNKNOWN))?;
            v.compression = Some(c.as_bytes().into());
        } else if args::is(key, b"training_threshold") && algo == Algo::Svs {
            training = Some(whole(label, key, val)?);
        } else if args::is(key, b"reduce") && algo == Algo::Svs {
            let _ = whole(label, key, val)?;
            reduce = true;
        } else {
            return Err(unwanted(label, key));
        }
    }

    // Two of the SVS options only mean something next to a compression, and one
    // of those only next to a compression of one kind. Both are checked after
    // the loop because the order the client wrote them in does not matter.
    if let Some(t) = training {
        if v.compression.is_none() {
            return Err(Fail::plain(NO_TRAINING));
        }
        if t < field::MIN_TRAINING {
            return Err(Fail::plain(SMALL_TRAINING));
        }
        v.training_threshold = Some(t);
    }
    if reduce && !v.compression.as_deref().is_some_and(is_leanvec) {
        return Err(Fail::plain(NO_REDUCE));
    }

    // All three are mandatory and a real server names the first one missing in
    // the order they are listed here.
    v.width = width.ok_or_else(|| missing(label, "TYPE"))?;
    v.dim = dim.ok_or_else(|| missing(label, "DIM"))?;
    v.metric = metric.ok_or_else(|| missing(label, "DISTANCE_METRIC"))?;
    Ok(v)
}

/// The compression schemes the Vamana form takes, in the spelling `FT.INFO`
/// gives back. The client's own case is not kept, because a real server takes
/// `lvq8` and reports it in the spelling it knows it by.
const COMPRESSIONS: &[&str] = &[
    "LVQ8",
    "LVQ4",
    "LVQ4x4",
    "LVQ4x8",
    "LeanVec4x8",
    "LeanVec8x8",
];

/// The scheme a word names, or `None` for a word that names none.
fn compression(v: &[u8]) -> Option<&'static str> {
    COMPRESSIONS
        .iter()
        .copied()
        .find(|c| args::is(v, c.as_bytes()))
}

/// Whether a scheme is one of the two that project the vector down first, which
/// are the only two `REDUCE` means anything beside.
fn is_leanvec(c: &[u8]) -> bool {
    c.starts_with(b"LeanVec")
}

/// `Bad arguments for algorithm HNSW: EF_RUNTIME`, which is the line for a
/// keyword that is real and belongs to another algorithm, and the line for a
/// keyword that is not real at all. A real server does not tell the two apart.
fn unwanted<'a>(label: &'static str, key: &'a [u8]) -> Fail<'a> {
    Fail::naming(
        match label {
            "FLAT" => "SEARCH_PARSE_ARGS Bad arguments for algorithm FLAT: ",
            "HNSW" => "SEARCH_PARSE_ARGS Bad arguments for algorithm HNSW: ",
            _ => "SEARCH_PARSE_ARGS Bad arguments for algorithm SVS-VAMANA: ",
        },
        key,
    )
}

/// `Bad arguments for vector similarity FLAT index `DIM`: why`.
fn vbad<'a>(label: &'static str, key: &'a [u8], why: &'static str) -> Fail<'a> {
    // The label and the key are two pieces and a line has room for one, so the
    // key is the piece that came from the client and the label is folded into
    // the head by the caller's choice of constant. There are three labels and
    // they are all `'static`, so this is a match rather than a format.
    Fail::about(
        match label {
            "FLAT" => "SEARCH_PARSE_ARGS Bad arguments for vector similarity FLAT index `",
            "HNSW" => "SEARCH_PARSE_ARGS Bad arguments for vector similarity HNSW index `",
            _ => "SEARCH_PARSE_ARGS Bad arguments for vector similarity SVS-VAMANA index `",
        },
        key,
        why,
    )
}

/// `Missing mandatory parameter: cannot create FLAT index without specifying
/// DIM argument`.
fn missing(label: &'static str, what: &'static str) -> Fail<'static> {
    Fail::about(
        match label {
            "FLAT" => {
                "SEARCH_PARSE_ARGS Missing mandatory parameter: cannot create FLAT index without specifying "
            }
            "HNSW" => {
                "SEARCH_PARSE_ARGS Missing mandatory parameter: cannot create HNSW index without specifying "
            }
            _ => {
                "SEARCH_PARSE_ARGS Missing mandatory parameter: cannot create SVS-VAMANA index without specifying "
            }
        },
        what.as_bytes(),
        " argument",
    )
}

/// A vector parameter that has to be a whole number.
fn whole<'a>(
    label: &'static str,
    key: &'a [u8],
    val: &[u8],
) -> core::result::Result<u64, Fail<'a>> {
    let n = parse_i64(val).ok_or_else(|| vbad(label, key, V_NOT_A_NUMBER))?;
    u64::try_from(n).map_err(|_| vbad(label, key, V_OUT_OF_RANGE))
}

/// How wide one coordinate is, from the word after `TYPE`.
fn width(v: &[u8]) -> Option<Width> {
    [
        Width::Int8,
        Width::Uint8,
        Width::Float16,
        Width::BFloat16,
        Width::Float32,
        Width::Float64,
    ]
    .into_iter()
    .find(|w| args::is(v, w.token().as_bytes()))
}

/// What the index measures, from the word after `DISTANCE_METRIC`.
fn metric(v: &[u8]) -> Option<Metric> {
    if args::is(v, b"l2") {
        Some(Metric::L2)
    } else if args::is(v, b"ip") {
        Some(Metric::Ip)
    } else if args::is(v, b"cosine") {
        Some(Metric::Cosine)
    } else {
        None
    }
}

/// `FT.ALTER index [SKIPINITIALSCAN] SCHEMA ADD field type ...`
///
/// Several fields at once, which is what the grammar allows even though the
/// documentation shows one.
fn alter<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out, ifnx: bool) -> Answer<'a> {
    let name = args.get(1);
    // The name is resolved before the grammar after it is read, which is why a
    // call that goes on to be refused has still counted as a use of the index.
    // The lookup does not get to decide the answer though, only to count: a name
    // that is not there next to a word that is not SCHEMA still answers about the
    // word.
    reg.touch(name);
    let mut at = 2;
    if args
        .opt(at)
        .is_some_and(|a| args::is(a, b"skipinitialscan"))
    {
        at += 1;
    }
    if !args.opt(at).is_some_and(|a| args::is(a, b"schema")) {
        return Err(Fail::plain(AFTER_ALTER));
    }
    at += 1;
    if !args.opt(at).is_some_and(|a| args::is(a, b"add")) {
        return Err(Fail::plain(ALTER_ACTION));
    }
    at += 1;

    // The index is looked up after the shape of the command is checked, which
    // is a real server's order: `FT.ALTER nope BLAH ADD q NUMERIC` answers that
    // ALTER must be followed by SCHEMA rather than that there is no such index.
    let Some(index) = reg.get(name) else {
        return Err(Fail::naming(MISSING, name));
    };
    let mut schema = index.schema.clone();
    match fields(args, at, &mut schema) {
        Ok(()) => {}
        // The quiet form swallows a field that is already there and nothing
        // else, so a schema with a bad type in it is still an error.
        Err(f) if ifnx && f.head == DUPLICATE => {
            out.ok();
            return Ok(());
        }
        Err(f) => return Err(f),
    }
    if let Some(index) = reg.get_mut(name) {
        index.schema = schema;
    }
    out.ok();
    Ok(())
}

/// `FT.DROPINDEX index [DD]`, and the three other spellings of it.
///
/// The two spellings take opposite defaults and each takes only its own word,
/// which is measured and is the whole reason this is one function with a flag
/// rather than two. `FT.DROPINDEX` keeps the documents and deletes them when it
/// is given `DD`, and `FT.DROP` deletes them and keeps them when it is given
/// `KEEPDOCS`. The word the other one takes is an unknown argument on both:
/// `FT.DROP i DD` is refused and so is `FT.DROPINDEX i KEEPDOCS`.
///
/// What comes back is the keys to delete, which are the ones the index actually
/// read and not the ones its prefix would have covered. A plain string under the
/// same prefix was never a document, so it stays. They are deleted above rather
/// than here because the registry is held and a stripe cannot be taken under it.
///
/// The index is looked up before the arguments after it are, so
/// `FT.DROP nope junk` answers that there is no such index and `FT.DROP i junk`
/// answers that `junk` is not an argument.
fn drop_index<'a>(
    reg: &mut Registry,
    spec: &Spec,
    args: Args<'a>,
    out: &mut Out,
    ifx: bool,
    newer: bool,
) -> core::result::Result<Option<After<'a>>, Fail<'a>> {
    // All four spellings count their own arguments, which is why the table
    // cannot do it for them. Which name goes in the line depends on which end
    // the count went wrong at: too few names the command plainly and too many
    // names it with an underscore in front, because the plain form is the one a
    // client called and the underscore form is the one the coordinator hands the
    // longer call on to. That is not a typo, it is the module's registration
    // showing through, and a client that matches on the whole line sees it.
    if args.len() < 2 {
        return Err(Fail::about(ARITY_PLAIN, spec.name.as_bytes(), ARITY_END));
    }
    if args.len() > 3 {
        return Err(Fail::about(ARITY, spec.name.as_bytes(), ARITY_END));
    }
    let name = args.get(1);
    if !reg.touch(name) {
        if ifx {
            out.ok();
            return Ok(None);
        }
        return Err(Fail::naming(MISSING, name));
    }
    let word = match newer {
        true => &b"dd"[..],
        false => &b"keepdocs"[..],
    };
    let said = match args.opt(2) {
        Some(a) if args::is(a, word) => true,
        Some(_) => return Err(Fail::plain(UNKNOWN_BARE)),
        None => false,
    };
    let index = reg.drop(name);
    out.ok();
    // The newer spelling deletes when it was asked to and the older one deletes
    // unless it was asked not to, so the word means the same thing on both and
    // the default does not.
    if said != newer {
        return Ok(None);
    }
    Ok(index.ok().map(|index| {
        After::Sweep(
            index
                .held
                .docs
                .all()
                .map(|(_, doc)| doc.key.clone())
                .collect(),
        )
    }))
}

/// `FT._LIST`, every index by name.
///
/// A set on RESP3 and an array on RESP2, of simple strings on both, which is
/// not the shape `FT.ALIASLIST` answers with even though the two look like the
/// same question asked twice.
///
/// One argument after the name is taken and ignored and two are too many, which
/// is the arity a real server enforces even though `COMMAND INFO` reports this
/// as taking any number at all.
fn list<'a>(reg: &Registry, spec: &Spec, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    if args.len() > 2 {
        return Err(Fail::about(ARITY_PLAIN, spec.name.as_bytes(), ARITY_END));
    }
    out.set(reg.len());
    for index in reg.iter() {
        word(out, &index.name);
    }
    Ok(())
}

/// `FT.ALIASADD alias index`.
///
/// The suffix on `FT._ALIASADDIFNX` forgives less than the name suggests. It
/// forgives the one case where the alias is already pointing at the index the
/// client is asking for, which is the call that has nothing left to do, and it
/// refuses an alias pointing at another index just as loudly as the plain form
/// does. An index that is not there is refused by both, and the two do not say
/// it the same way: the plain form answers the not found line the rest of the
/// group uses and the suffixed one answers the line about a name that might be
/// an alias, which is the wrapper around it doing its own lookup first. An index
/// name that is really an alias gets that longer line from both of them, because
/// this is the one command in the group that will not follow an alias and so the
/// one place where the difference between a name and an alias is worth a
/// sentence of its own.
fn alias_add<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out, ifnx: bool) -> Answer<'a> {
    let alias = args.get(1);
    let name = args.get(2);
    // The index on the right is looked up first and counts as a use of it, even
    // on the calls that go on to refuse the alias on the left. It is looked up
    // by its own name and never through an alias, which is why the count only
    // moves for a real index name.
    if reg.named(name).is_some() {
        reg.touch(name);
    }
    if ifnx
        && let Some(at) = reg.target(alias)
        && reg.named(name).is_some_and(|i| *i.name == *at)
    {
        out.ok();
        return Ok(());
    }
    match reg.alias(alias, name) {
        Ok(()) => out.ok(),
        Err(Clash::IsIndex) => return Err(Fail::plain(CONFLICT)),
        Err(Clash::Aliased) => return Err(Fail::plain(ALIAS_EXISTS)),
        // A target that is itself an alias reads the same from both spellings,
        // and it is the only case where the plain form says the longer sentence.
        Err(Clash::IsAlias) => return Err(Fail::plain(NO_TARGET)),
        Err(_) if ifnx => return Err(Fail::plain(NO_TARGET)),
        Err(_) => return Err(Fail::naming(MISSING, name)),
    }
    Ok(())
}

/// `FT.ALIASUPDATE alias index`, which moves an alias that is already pointing
/// somewhere and adds one that is not.
fn alias_update<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let alias = args.get(1);
    let name = args.get(2);
    // Where the alias was pointing, read before the move takes it away.
    let old: Option<Box<[u8]>> = reg.target(alias).map(Into::into);
    match reg.realias(alias, name) {
        Ok(()) => {
            // Both ends count, so moving an alias from one index to another
            // counts a use on the one it left as well as on the one it landed
            // on. Only a move that happens counts: this is the one command in
            // the group that counts nothing at all when it refuses, where the
            // rest have all counted by the time they work out the answer.
            if let Some(old) = old {
                reg.touch(&old);
            }
            reg.touch(name);
            out.ok();
        }
        // Not the conflict `FT.ALIASADD` answers with for the same argument,
        // which is a real server's inconsistency and is copied because a client
        // branches on the code word in front of it.
        Err(Clash::IsIndex) => return Err(Fail::plain(NOT_MINE)),
        Err(Clash::IsAlias) => return Err(Fail::plain(NO_TARGET)),
        Err(_) => return Err(Fail::naming(MISSING, name)),
    }
    Ok(())
}

/// `FT.ALIASDEL alias`.
///
/// The name of an index is refused by both forms, the one with the suffix
/// included, and with the sentence about ownership rather than the one about a
/// missing alias. The suffix only forgives a name that is nothing at all, and an
/// index name is something, it is just not an alias.
fn alias_del<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out, ifx: bool) -> Answer<'a> {
    let alias = args.get(1);
    // The suffixed form resolves the name itself before it hands the call on to
    // the plain one, which then resolves it again, so a name that resolves at all
    // counts twice under the suffixed form and once under the plain one.
    if ifx {
        reg.touch(alias);
    }
    reg.touch(alias);
    if reg.named(alias).is_some() {
        return Err(Fail::plain(NOT_MINE));
    }
    match reg.unalias(alias) {
        Ok(()) => out.ok(),
        Err(_) if ifx => out.ok(),
        Err(_) => return Err(Fail::plain(NO_ALIAS)),
    }
    Ok(())
}

/// `FT.ALIASLIST index`, the aliases pointing at one index.
///
/// The argument is an index and not an alias, so asking this about an alias
/// answers that there is no index by that name.
fn alias_list<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    if reg.named(name).is_none() {
        return Err(Fail::naming(MISSING, name));
    }
    // Only an index name gets this far, and finding it counts as a use of it the
    // same way every other lookup by name does.
    let real: Box<[u8]> = name.into();
    reg.open(&real);
    // Bulk strings where `FT._LIST` writes simple ones, and a set on RESP3 the
    // same way.
    let n = reg.aliases_of(&real).count();
    out.set(n);
    for alias in reg.aliases_of(&real) {
        out.bulk(alias);
    }
    Ok(())
}

/// `FT.TAGVALS index attribute`, every distinct value a tag field holds.
///
/// The values come back as they are stored rather than as they arrived, which
/// for an ordinary tag field means folded and trimmed: `Red, BLUE ` written to
/// one document answers `blue` and `red`. A field declared `CASESENSITIVE`
/// keeps what it was given, so `Aa|bB` answers `Aa` and `bB`. They are sorted
/// by their bytes either way, which puts a capital in front of a small letter.
///
/// The name is the attribute and never the identifier, so a field declared
/// `AS gg` is asked about by `gg` and asking about `g` answers that there is no
/// such field. Looking up the index counts as a use of it on every road out of
/// here, the two that refuse the field included.
fn tag_vals<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    let attribute = args.get(2);
    let Some(index) = reg.open(name) else {
        return Err(Fail::naming(MISSING, name));
    };
    match index.field(attribute) {
        None => return Err(Fail::plain(NO_FIELD)),
        Some(field) if !matches!(field.kind, Kind::Tag(_)) => {
            return Err(Fail::plain(NOT_A_TAG));
        }
        Some(_) => {}
    }
    // A tag field nothing has been written to has no list at all, which answers
    // the same empty set as a list that has been emptied.
    let values = index.held.values(attribute);
    out.set(values.map_or(0, Tags::len));
    for (value, _) in values.into_iter().flat_map(Tags::all) {
        out.bulk(value);
    }
    Ok(())
}

/// `FT.DICTADD dict term [term ...]`, and how many of the terms were new.
///
/// The dictionary is made by the first word that goes into it and there is no
/// command to make one, so this is also the create. Nothing here touches an
/// index, so a dictionary named after one is not attached to it.
fn dict_add<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    let terms: Vec<&[u8]> = (2..args.len()).map(|i| args.get(i)).collect();
    out.int(reg.dicts.add(name, &terms) as i64);
    Ok(())
}

/// `FT.DICTDEL dict term [term ...]`, and how many of the terms were there.
///
/// A dictionary that does not exist is not an error, it is zero terms deleted,
/// and so is a term the dictionary does not hold.
fn dict_del<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    let terms: Vec<&[u8]> = (2..args.len()).map(|i| args.get(i)).collect();
    out.int(reg.dicts.del(name, &terms) as i64);
    Ok(())
}

/// `FT.DICTDUMP dict`, every term in it in byte order.
///
/// A set on RESP3 and an array of bulk strings on RESP2, and a dictionary
/// nobody ever made dumps empty rather than complaining, which is the one place
/// in this group where a missing name is not an error.
fn dict_dump<'a>(reg: &Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    out.set(reg.dicts.len(name));
    for term in reg.dicts.dump(name) {
        out.bulk(term);
    }
    Ok(())
}

/// `FT.SYNUPDATE index group [SKIPINITIALSCAN] term [term ...]`
///
/// Every term joins the group, and a term already in another group joins this
/// one as well rather than moving. The terms are folded on the way in, which is
/// what the dump answers with, and the group id is not: `FT.SYNUPDATE i G1 Boy`
/// dumps `boy` under `G1`.
///
/// A group means nothing to a document that was written before it, because a
/// group is written into the index at the same time the words are. So the index
/// is read again from the keyspace unless the client says `SKIPINITIALSCAN`,
/// and that renumbers every document in it. `SKIPINITIALSCAN` is only that word
/// in that one place: `FT.SYNUPDATE i g boy SKIPINITIALSCAN` puts the word
/// `skipinitialscan` in the group and reads the index again.
///
/// Looking up the index counts as a use of it, and a wrong arity is refused
/// before that happens, so the count moves for a command that ran and not for
/// one that never started.
fn syn_update<'a>(
    reg: &mut Registry,
    args: Args<'a>,
    out: &mut Out,
) -> core::result::Result<Option<After<'a>>, Fail<'a>> {
    let name = args.get(1);
    let group = args.get(2);
    let skip = args.opt(3).is_some_and(|a| args::is(a, b"skipinitialscan"));
    let from = if skip { 4 } else { 3 };
    let terms: Vec<&[u8]> = (from..args.len()).map(|i| args.get(i)).collect();
    let Some(index) = reg.open(name) else {
        return Err(Fail::naming(MISSING, name));
    };
    index.synonyms.update(group, &terms);
    out.ok();
    Ok((!skip).then_some(After::Scan(Fill { name, obeys: false })))
}

/// `FT.SYNDUMP index`, every term in a group and which groups it is in.
///
/// A map on RESP3 and a flat run of a term and its groups on RESP2. The terms
/// come back in byte order, where a real server answers in whatever order its
/// hash table holds them, which is D-87. The groups under one term are in the
/// order the term was added to them and that is the same order either way.
fn syn_dump<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    let Some(index) = reg.open(name) else {
        return Err(Fail::naming(MISSING, name));
    };
    out.map(index.synonyms.len());
    for (term, ids) in index.synonyms.dump() {
        out.bulk(term);
        out.array(ids.len());
        for id in ids {
            out.bulk(id);
        }
    }
    Ok(())
}

/// The word `FT.SPELLCHECK` puts in front of every row it answers with.
const TERM: &[u8] = b"TERM";
/// What a `RESP3` spellcheck reply calls the one thing it holds.
const RESULTS: &[u8] = b"results";
/// A distance that is not a whole number from one to four.
const BAD_DISTANCE: &str = "bad distance given, distance must be a natural number between 1 to 4";
/// `DISTANCE` with nothing after it.
const NEED_DISTANCE: &str = "DISTANCE arg is given but no DISTANCE comes after";
/// `TERMS` without both of the words it takes.
const NEED_TERMS: &str = "TERM arg is given but no TERM params comes after";
/// `TERMS` followed by something that is neither `INCLUDE` nor `EXCLUDE`.
const BAD_TERMS: &str = "bad format, exclude/include operation was not given";
/// A dictionary nobody has put a word in.
const NO_DICT: &str = "Dict does not exist: ";

/// What the arguments after the query asked for.
struct Spelling<'a> {
    /// How far a suggestion may be from the word.
    distance: u8,
    /// Whether a distance has already been read, since the first one counts and
    /// the rest are dropped.
    fixed: bool,
    /// The names of the dictionaries whose words are candidates.
    include: Vec<&'a [u8]>,
    /// The names of the dictionaries whose words are spelled right.
    exclude: Vec<&'a [u8]>,
    /// `DIALECT` and `PARAMS`, which arrive here the way they do everywhere
    /// else and are handed to the same parser.
    asked: Asked<'a>,
}

/// `FT.SPELLCHECK index query [DISTANCE n] [TERMS INCLUDE|EXCLUDE dict] ...`
///
/// The query is parsed the way `FT.SEARCH` would parse it and every plain word
/// in the tree is answered about, in the order it was written and twice when it
/// was written twice. A prefix, a suffix, an infix, a pattern and a fuzzy word
/// stand for terms the index already holds and are not words anybody could have
/// misspelled, so none of them is answered about.
///
/// The stopword list is off, because a real server checks `then` and suggests
/// `thin` for it. Nothing indexes a stopword, so a stopword is always a word the
/// index does not hold.
///
/// An argument nobody knows is stepped over rather than refused, which is not
/// how the rest of the search surface reads its arguments and is measured:
/// `FT.SPELLCHECK i q BOGUS` answers. A second `DISTANCE` is dropped the same
/// way, so `DISTANCE 3 DISTANCE 1` looks three letters out.
fn spellcheck<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    let query = args.get(2);
    // Opened rather than named, because a spellcheck counts as a use of the
    // index the same way a search does. The index itself is let go of straight
    // away and taken again below, because the dictionaries and the indexes are
    // both on the registry and the check needs to read the two of them at once.
    if reg.open(name).is_none() {
        return Err(Fail::naming(MISSING, name));
    }
    let asked = match spelling(args, 3) {
        Ok(asked) => asked,
        Err(text) => {
            out.error(&text);
            return Ok(());
        }
    };
    for named in asked.include.iter().chain(&asked.exclude) {
        if reg.dicts.len(named) == 0 {
            out.error(&line(NO_DICT, named, ""));
            return Ok(());
        }
    }
    let lists = Lists {
        include: asked
            .include
            .iter()
            .flat_map(|n| reg.dicts.dump(n))
            .collect(),
        exclude: asked
            .exclude
            .iter()
            .flat_map(|n| reg.dicts.dump(n))
            .collect(),
    };
    let index = reg.named(name).expect("an index that was just opened");
    let ask = Ask {
        dialect: asked.asked.dialect,
        params: &asked.asked.params,
        verbatim: false,
        // A real server checks a stopword and suggests for it, which it can
        // only do by reading the query with the list turned off.
        stopwords: false,
    };
    let node = match query::parse(query, index, &ask) {
        Ok(node) => node,
        Err(bad) => {
            out.error(&refused(&bad));
            return Ok(());
        }
    };
    let found = spell::check(index, &node, asked.distance, &lists);
    if out.proto().is_resp3() {
        out.map(1);
        out.bulk(RESULTS);
        out.map(found.len());
        for checked in &found {
            out.bulk(&checked.word);
            out.array(checked.guesses.len());
            for guess in &checked.guesses {
                out.map(1);
                out.bulk(&guess.term);
                out.double(guess.score);
            }
        }
        return Ok(());
    }
    out.array(found.len());
    for checked in &found {
        out.array(3);
        out.bulk(TERM);
        out.bulk(&checked.word);
        out.array(checked.guesses.len());
        for guess in &checked.guesses {
            out.array(2);
            out.double(guess.score);
            out.bulk(&guess.term);
        }
    }
    Ok(())
}

/// The words after the query, which are read leniently on this one command.
fn spelling(args: Args<'_>, from: usize) -> core::result::Result<Spelling<'_>, Vec<u8>> {
    let mut spelling = Spelling {
        distance: spell::NEAREST,
        fixed: false,
        include: Vec::new(),
        exclude: Vec::new(),
        asked: Asked::default(),
    };
    let mut at = from;
    while at < args.len() {
        let word = args.get(at);
        if args::is(word, b"DISTANCE") {
            let Some(value) = args.opt(at + 1) else {
                return Err(NEED_DISTANCE.as_bytes().to_vec());
            };
            let Some(distance) = parse_i64(value)
                .filter(|d| (i64::from(spell::NEAREST)..=i64::from(spell::FURTHEST)).contains(d))
            else {
                return Err(BAD_DISTANCE.as_bytes().to_vec());
            };
            if !spelling.fixed {
                spelling.distance = u8::try_from(distance).unwrap_or(spell::NEAREST);
                spelling.fixed = true;
            }
            at += 2;
            continue;
        }
        if args::is(word, b"TERMS") {
            let (Some(how), Some(named)) = (args.opt(at + 1), args.opt(at + 2)) else {
                return Err(NEED_TERMS.as_bytes().to_vec());
            };
            if args::is(how, b"INCLUDE") {
                spelling.include.push(named);
            } else if args::is(how, b"EXCLUDE") {
                spelling.exclude.push(named);
            } else {
                return Err(BAD_TERMS.as_bytes().to_vec());
            }
            at += 3;
            continue;
        }
        if args::is(word, b"DIALECT") {
            let Some(value) = args.opt(at + 1) else {
                return Err(line(NEED_ARG, b"DIALECT", ""));
            };
            let Some(dialect) = parse_i64(value).filter(|d| (1..=i64::from(NEWEST)).contains(d))
            else {
                return Err(BAD_DIALECT.as_bytes().to_vec());
            };
            spelling.asked.dialect = u8::try_from(dialect).unwrap_or(1);
            at += 2;
            continue;
        }
        if args::is(word, b"PARAMS") {
            at = params(args, at, &mut spelling.asked)?;
            continue;
        }
        // Everything else is stepped over. A real server reads this list with a
        // loop that only knows those four words and leaves the rest alone, so
        // `VERBATIM` here is not an error and is not honoured either.
        at += 1;
    }
    Ok(spelling)
}

/// A string in a reply, as a simple string when it can be one.
///
/// `FT.INFO` writes nearly everything as a simple string, the index name and
/// the key prefixes included, and those are the client's own bytes. A real
/// server writes them straight out and a prefix with a newline in it breaks the
/// stream it goes into. A bulk string carries the same bytes and says how many
/// there are, so the value that cannot be a simple string goes out as one and
/// every ordinary value is byte for byte what a real server sends.
fn word(out: &mut Out, s: &[u8]) {
    if s.contains(&b'\r') || s.contains(&b'\n') {
        out.bulk(s);
    } else {
        out.simple(s);
    }
}

/// A name and a simple string value.
fn pair(out: &mut Out, name: &str, value: &[u8]) {
    out.simple(name.as_bytes());
    word(out, value);
}

/// A name and a number that is written as a double on RESP3 and as a bulk
/// string of the same digits on RESP2.
fn number(out: &mut Out, name: &str, value: f64) {
    out.simple(name.as_bytes());
    out.double(value);
}

/// A name and a count, which is an integer on both protocols.
fn tally(out: &mut Out, name: &str, value: u64) {
    out.simple(name.as_bytes());
    out.uint(value);
}

/// `FT.INFO index`, everything the server knows about one index.
fn info<'a>(server: &Server, reg: &mut Registry, args: Args<'a>, out: &mut Out) -> Answer<'a> {
    let name = args.get(1);
    // Counted before it is reported, so the first `FT.INFO` after a create says
    // one rather than nought. An alias counts once and not twice, because the
    // alias is resolved on the way to the one lookup rather than being a lookup
    // of its own.
    let Some(index) = reg.open(name) else {
        return Err(Fail::naming(MISSING, name));
    };

    // A cursor nobody is reading is a cursor nobody has read from since the
    // command that made it finished, so the two global numbers are the same one.
    let (whole, mine) = cursor::stats(server, &index.name);
    let idle = whole;

    let d = &index.definition;
    let stopwords = d.stopwords.as_ref();
    out.map(34 + usize::from(stopwords.is_some()));

    pair(out, "index_name", &index.name);
    out.simple(b"index_options");
    let tokens = d.options.tokens();
    out.array(tokens.len());
    for t in &tokens {
        out.simple(t.as_bytes());
    }

    out.simple(b"index_definition");
    let mut pairs = 4;
    pairs += usize::from(d.filter.is_some());
    pairs += usize::from(d.language.is_some());
    pairs += usize::from(d.language_field.is_some());
    pairs += usize::from(d.score_field.is_some());
    pairs += usize::from(d.payload_field.is_some());
    out.map(pairs);
    pair(out, "key_type", d.on.token().as_bytes());
    out.simple(b"prefixes");
    out.array(d.prefixes.len());
    for p in &d.prefixes {
        word(out, p);
    }
    if let Some(f) = &d.filter {
        pair(out, "filter", f);
    }
    if let Some(l) = &d.language {
        pair(out, "default_language", l);
    }
    if let Some(l) = &d.language_field {
        pair(out, "language_field", l);
    }
    number(out, "default_score", d.score);
    if let Some(s) = &d.score_field {
        pair(out, "score_field", s);
    }
    if let Some(p) = &d.payload_field {
        pair(out, "payload_field", p);
    }
    // Always false, on a real server as well as here. Every index that ought to
    // report true reports false there, including one created with an explicit
    // empty prefix and no filter, so this is measured rather than computed.
    pair(out, "indexes_all", b"false");

    out.simple(b"attributes");
    out.array(index.schema.len());
    for f in &index.schema {
        attribute(out, f);
    }

    // Nothing is indexed yet, so every one of these is nought rather than a
    // number this server has no way to produce. D-58 is the register entry.
    let docs = index.held.docs.len() as u64;
    let records = index.held.records();
    tally(out, "num_docs", docs);
    tally(out, "max_doc_id", u64::from(index.held.docs.last()));
    tally(out, "num_terms", index.held.words() as u64);
    tally(out, "num_records", records);
    number(out, "inverted_sz_mb", 0.0);
    number(out, "vector_index_sz_mb", 0.0);
    tally(out, "total_inverted_index_blocks", 0);
    number(out, "offset_vectors_sz_mb", 0.0);
    number(out, "doc_table_size_mb", 0.0);
    number(out, "sortable_values_size_mb", 0.0);
    number(out, "key_table_size_mb", 0.0);
    number(out, "tag_overhead_sz_mb", 0.0);
    number(out, "text_overhead_sz_mb", 0.0);
    number(out, "total_index_memory_sz_mb", 0.0);
    number(out, "geoshapes_sz_mb", 0.0);
    // Four averages, and the three after the first are over byte counts this
    // build does not keep yet. All four are a division by nought on an empty
    // index, and a real server answers `nan` there for the same reason rather
    // than as a placeholder.
    //
    // The division is done in single precision and printed in double, which is
    // what a real server does and is visible: seventeen records over three
    // documents comes out as 5.666666507720947 and not 5.666666666666667. The
    // counters are doubles by the time they reach the reply and the arithmetic
    // behind them is not, so a client comparing the two builds sees the same
    // digits.
    number(
        out,
        "records_per_doc_avg",
        f64::from(records as f32 / docs as f32),
    );
    number(out, "bytes_per_record_avg", f64::NAN);
    number(out, "offsets_per_term_avg", f64::NAN);
    number(out, "offset_bits_per_record_avg", f64::NAN);
    tally(
        out,
        "hash_indexing_failures",
        index.trouble.whole().failures(),
    );
    number(out, "total_indexing_time", 0.0);
    tally(out, "indexing", 0);
    number(out, "percent_indexed", 1.0);
    tally(out, "number_of_uses", index.uses);
    tally(out, "cleaning", 0);

    out.simple(b"gc_stats");
    out.map(7);
    number(out, "bytes_collected", 0.0);
    number(out, "total_ms_run", 0.0);
    number(out, "total_cycles", 0.0);
    number(out, "average_cycle_time_ms", f64::NAN);
    number(out, "last_run_time_ms", 0.0);
    number(out, "gc_numeric_trees_missed", 0.0);
    number(out, "gc_blocks_denied", 0.0);

    out.simple(b"cursor_stats");
    out.map(4);
    tally(out, "global_idle", idle);
    tally(out, "global_total", whole);
    tally(out, "index_capacity", CURSOR_CAPACITY);
    tally(out, "index_total", mine);

    if let Some(list) = stopwords {
        out.simple(b"stopwords_list");
        out.array(list.len());
        for w in list {
            out.bulk(w);
        }
    }

    out.simple(b"dialect_stats");
    out.map(4);
    for n in 1..=4 {
        out.simple(match n {
            1 => b"dialect_1".as_slice(),
            2 => b"dialect_2",
            3 => b"dialect_3",
            _ => b"dialect_4",
        });
        out.uint(0);
    }

    out.simple(b"Index Errors");
    out.map(4);
    errors(out, index.trouble.whole());
    out.simple(b"background indexing status");
    out.simple(b"OK");

    out.simple(b"field statistics");
    out.array(index.schema.len());
    for f in &index.schema {
        statistics(out, f, index.trouble.field(&f.attribute));
    }
    Ok(())
}

/// How many cursors an index has room for, which is what `FT.INFO` reports
/// before a single one has been opened.
const CURSOR_CAPACITY: u64 = 128;

/// One schema field, as `FT.INFO` describes it.
///
/// The order is fixed and is not the order the client wrote the options in: a
/// field declared `SORTABLE NOSTEM` and one declared `NOSTEM SORTABLE` are one
/// field and describe themselves the same way.
///
/// The two protocols disagree about the shape. RESP2 writes a flat array with
/// the flag words on the end of it, and RESP3 writes a map with the flags
/// gathered into an array under `flags`, which is there even when it is empty.
fn attribute(out: &mut Out, f: &Field) {
    let mut flags: Vec<&str> = Vec::new();
    if f.sortable {
        flags.push("SORTABLE");
    }
    if f.is_unf() {
        flags.push("UNF");
    }
    if let Kind::Text(t) = &f.kind
        && t.nostem
    {
        flags.push("NOSTEM");
    }
    if let Kind::Tag(t) = &f.kind
        && t.casesensitive
    {
        flags.push("CASESENSITIVE");
    }
    if f.suffix_trie {
        flags.push("WITHSUFFIXTRIE");
    }
    if f.index_empty {
        flags.push("INDEXEMPTY");
    }
    if f.index_missing {
        flags.push("INDEXMISSING");
    }
    if f.noindex {
        flags.push("NOINDEX");
    }

    // A tag reports its separator and a text its weight where the flags go, in
    // front of them, so the pairs are counted before the shape is chosen.
    let pairs = 3 + match &f.kind {
        Kind::Text(_) | Kind::Tag(_) => 1,
        Kind::GeoShape(_) => 1,
        Kind::Vector(v) => match v.algo {
            Algo::Flat => 4,
            Algo::Hnsw => 7,
            // The Vamana form reports what it compresses to always and how many
            // vectors it trains that compression over only when there is one to
            // train, so a compressed field carries one pair more than a plain
            // one does.
            Algo::Svs => 7 + usize::from(v.compression.is_some()),
        },
        Kind::Numeric | Kind::Geo => 0,
    };
    if out.proto().is_resp3() {
        out.map(pairs + 1);
    } else {
        out.array(pairs * 2 + flags.len());
    }

    pair(out, "identifier", &f.identifier);
    pair(out, "attribute", &f.attribute);
    pair(out, "type", f.kind.token().as_bytes());
    match &f.kind {
        Kind::Text(t) => number(out, "WEIGHT", t.weight),
        Kind::Tag(t) => {
            out.simple(b"SEPARATOR");
            word(out, &[t.separator]);
        }
        Kind::GeoShape(c) => pair(out, "coord_system", c.token().as_bytes()),
        Kind::Vector(v) => {
            pair(out, "algorithm", v.algo.token().as_bytes());
            pair(out, "data_type", v.width.token().as_bytes());
            tally(out, "dim", v.dim);
            pair(out, "distance_metric", v.metric_token().as_bytes());
            match v.algo {
                Algo::Flat => {}
                Algo::Hnsw => {
                    tally(out, "M", v.m);
                    tally(out, "ef_construction", v.ef_construction);
                    tally(out, "ef_runtime", v.ef_runtime);
                }
                Algo::Svs => {
                    tally(out, "graph_max_degree", v.graph_max_degree);
                    tally(out, "construction_window_size", v.construction_window);
                    pair(
                        out,
                        "compression",
                        v.compression
                            .as_deref()
                            .unwrap_or(field::NO_COMPRESSION.as_bytes()),
                    );
                    if v.compression.is_some() {
                        tally(
                            out,
                            "training_threshold",
                            v.training_threshold.unwrap_or(field::TRAINING_THRESHOLD),
                        );
                    }
                }
            }
        }
        Kind::Numeric | Kind::Geo => {}
    }

    if out.proto().is_resp3() {
        out.simple(b"flags");
        out.array(flags.len());
    }
    for flag in &flags {
        out.simple(flag.as_bytes());
    }
}

/// The three lines an error block starts with, which are the same for the index
/// and for each of its fields.
///
/// The sentence is a simple string and the key is a bulk, which is not a
/// consistent pair and is what a real server writes, `N/A` included.
fn errors(out: &mut Out, e: &Errors) {
    tally(out, "indexing failures", e.failures());
    out.simple(b"last indexing error");
    out.simple(e.sentence());
    out.simple(b"last indexing error key");
    out.bulk(e.about());
}

/// One field's own error counters, which are all nought until a key it could
/// not read has been written.
///
/// A vector field carries four more than the rest, and they are the four a
/// client watching an index fill up would read.
fn statistics(out: &mut Out, f: &Field, e: &Errors) {
    let vector = matches!(f.kind, Kind::Vector(_));
    out.map(3 + if vector { 4 } else { 0 });
    pair(out, "identifier", &f.identifier);
    pair(out, "attribute", &f.attribute);
    out.simple(b"Index Errors");
    out.map(3);
    errors(out, e);
    if vector {
        tally(out, "memory", 0);
        tally(out, "marked_deleted", 0);
        tally(out, "direct_hnsw_insertions", 0);
        tally(out, "flat_buffer_size", 0);
    }
}

/// The most recent grammar a client may ask for.
const NEWEST: u8 = 4;

const BAD_DIALECT: &str = "SEARCH_PARSE_ARGS DIALECT requires a non negative integer >=1 and <= 4";
const NEED_ARG: &str = "SEARCH_PARSE_ARGS Need an argument for ";
const ODD_PARAMS: &str = "SEARCH_ADD_ARGS Parameters must be specified in PARAM VALUE pairs";
const NOT_MAIN: &str = "` at position ";
const NOT_MAIN_END: &str = " for <main>";

/// The most rows one `FT.SEARCH` will hand back.
const MOST: i64 = 1_000_000;

const LIMIT_TWO: &str = "SEARCH_PARSE_ARGS LIMIT requires two arguments";
const LIMIT_NUMBERS: &str = "SEARCH_PARSE_ARGS LIMIT needs two numeric arguments";
const LIMIT_OVER: &str = "SEARCH_LIMIT_OVER LIMIT exceeds maximum of 1000000";
const LIMIT_START: &str = "SEARCH_LIMIT_OVER The `offset` of the LIMIT must be 0 when `num` is 0";
const TIMEOUT_ARG: &str = "SEARCH_PARSE_ARGS Need argument for TIMEOUT";
const TIMEOUT_NUMBER: &str = "SEARCH_PARSE_ARGS TIMEOUT requires a non negative integer";
const NO_SCORER: &str = "SEARCH_QUERY_BAD No such scorer ";
const NO_LANGUAGE: &str = "SEARCH_QUERY_BAD No such language";
const LOW_RANGE: &str = "SEARCH_PARSE_ARGS Bad lower range: ";
const HIGH_RANGE: &str = "SEARCH_PARSE_ARGS Bad upper range: ";
const BACKWARDS: &str = "SEARCH_SYNTAX Invalid numeric range (min > max): @";
const FILTER_THREE: &str = "SEARCH_PARSE_ARGS FILTER requires 3 arguments";
const GEO_FIVE: &str = "SEARCH_PARSE_ARGS GEOFILTER requires 5 arguments";
const GEO_UNIT: &str = "SEARCH_PARSE_ARGS Unknown distance unit ";
/// The code word in front of the two geo lines the parser also sends, which
/// reach the wire from here when the filter was written as an option rather
/// than inside the query.
const SYNTAX: &str = "SEARCH_SYNTAX ";
const NEED_NAME: &str = "SEARCH_PARSE_ARGS RETURN path AS name - must be accompanied with NAME";
const NEED_LOAD_NAME: &str = "SEARCH_PARSE_ARGS LOAD path AS name - must be accompanied with NAME";
const LOAD_COUNT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for LOAD: Expected number of fields or `*`";
const LOAD_BOUNDS: &str =
    "SEARCH_PARSE_ARGS Bad arguments for LOAD: Value is outside acceptable bounds";
const LOAD_SHORT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for LOAD: Expected an argument, but none provided";
const NOT_HERE: &str = " is not supported on FT.AGGREGATE";
const GROUP_SHORT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for GROUPBY: Expected an argument, but none provided";
const GROUP_COUNT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for GROUPBY: Could not convert argument to expected type";
/// What a `GROUPBY` says about a property written without its `@`, which is the
/// one place in the group that spells out what the client should have sent.
const NO_AT: &str = "SEARCH_PARSE_ARGS Bad arguments for GROUPBY: Unknown property `";
const NO_AT_MID: &str = "`. Did you mean `@";
const NO_AT_END: &str = "`?";
const SORT_SHORT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for SORTBY: Expected an argument, but none provided";
const SORT_COUNT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for SORTBY: Could not convert argument to expected type";
const SORT_BOUNDS: &str =
    "SEARCH_PARSE_ARGS Bad arguments for SORTBY: Value is outside acceptable bounds";
const MAX_COUNT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for MAX: Could not convert argument to expected type";
/// What a word inside a `SORTBY` list that is neither a property nor a
/// direction answers, which names the word in brackets rather than in quotes.
const SORT_WAY: &str = "SEARCH_PARSE_ARGS MISSING ASC or DESC after sort field (";
const SORT_WAY_END: &str = ")";
/// What a `SORTBY` says about a property it cannot find, which is worded its
/// own way and not the way an `APPLY` or a `FILTER` words the same thing.
const SORT_PROP: &str = "SEARCH_PROP_NOT_FOUND Property `";
const SORT_PROP_END: &str = "` not loaded nor in schema";
const SORT_TWICE: &str = "SEARCH_PARSE_ARGS Multiple SORTBY steps are not allowed. Sort multiple fields in a single step";
/// The same thing on a search, which says the first half of it and stops. The
/// advice about sorting several fields at once belongs to the step that can do
/// that, and a search cannot: it sorts by one field or by none.
const SORT_ONCE: &str = "SEARCH_PARSE_ARGS Multiple SORTBY steps are not allowed";
/// What a `SORTBY` with no field after it answers on a search, which is not the
/// line the same mistake gets on an aggregation.
const SORT_BARE: &str = "SEARCH_PARSE_ARGS Bad SORTBY arguments";
/// A search sorts the whole answer and hands back a window on it, so the cap a
/// sort step takes on an aggregation has nothing to do here. The word is
/// recognised anyway rather than being left to the unknown argument line.
const SORT_MAX: &str = "SEARCH_PARSE_ARGS SORTBY MAX is not supported by FT.SEARCH";
/// What `EXPLAINSCORE` answers when nothing asked for the scores it explains.
const SCORE_ALONE: &str = "SEARCH_PARSE_ARGS EXPLAINSCORE must be accompanied with WITHSCORES";
const NO_PROPERTY: &str = "SEARCH_PROP_NOT_FOUND No such property `";
const NOT_LOADED: &str = "SEARCH_PROP_NOT_FOUND Property not loaded nor in pipeline: `";
const QUOTE_END: &str = "`";
const DUPLICATE_PROP: &str = "SEARCH_FIELD_DUP Property `";
const DUPLICATE_END: &str = "` specified more than once";
const LOAD_LATE: &str = "SEARCH_QUERY_BAD LOAD cannot be applied after projectors or reducers";
/// What an `APPLY` or a `FILTER` with no expression after it answers, which
/// names the pair of them rather than the one the client wrote.
const STEP_SHORT: &str =
    "SEARCH_PARSE_ARGS Bad arguments for APPLY/FILTER: Expected an argument, but none provided";
const AS_SHORT: &str = "SEARCH_PARSE_ARGS AS needs argument";
/// What a bare `REDUCE` answers, which names the reason for the failure as the
/// word for no failure at all. That is what a real server sends.
const REDUCE_BARE: &str = "SEARCH_PARSE_ARGS Bad arguments for REDUCE: SUCCESS";
const NO_REDUCER: &str = "SEARCH_REDUCER_NOT_FOUND No such reducer: ";
const MISSING_ARGS: &str = "SEARCH_PARSE_ARGS Missing arguments for ";
const COUNT_ONLY: &str = "SEARCH_ATTR_BAD Count accepts 0 values only";
const PERCENTAGE: &str = "SEARCH_PARSE_ARGS Percentage must be between 0.0 and 1.0";
const RESOLUTION: &str = "SEARCH_PARSE_ARGS Invalid resolution";
const SAMPLE_BIG: &str = "SEARCH_PARSE_ARGS Sample size too large";
/// The two reducer arguments a real server names in angle brackets rather than
/// by the keyword they follow, because neither of them follows a keyword.
const SAMPLE_SIZE: &str = "SEARCH_PARSE_ARGS Bad arguments for <sample size>";
const RESOLUTION_ARG: &str = "SEARCH_PARSE_ARGS Bad arguments for <resolution>";
/// The largest a `RANDOM_SAMPLE` and the finest a `QUANTILE` may ask for, both
/// measured: a thousand is taken and a thousand and one is refused.
const MOST_SAMPLE: i64 = 1000;

/// What a `RETURN` list came to: for each field, where its value is read from on
/// the key and what name it comes back under.
type Returns<'a> = Vec<(Box<[u8]>, &'a [u8])>;

/// What a client asked for about the rows, which only `FT.SEARCH` acts on.
///
/// `FT.EXPLAIN` reads the same list and throws this half of it away. It still
/// has to read it, because a real server refuses `LIMIT 0 -1` and
/// `SCORER NOPE` on either command: the argument list is checked once and what
/// happens to the result set afterwards is a separate question.
struct Rows<'a> {
    /// Whether the fields of each key come back, which `NOCONTENT` turns off
    /// and nothing turns back on. `RETURN 0` leaves this alone and empties the
    /// list instead, which is why the two are not one flag: a `RETURN 0` with
    /// another `RETURN` after it answers with fields and a `NOCONTENT` with a
    /// `RETURN` after it does not.
    content: bool,
    scores: bool,
    payloads: bool,
    /// Where the window starts and how wide it is, which is `LIMIT 0 10` when
    /// nobody said.
    offset: usize,
    count: usize,
    /// The fields to send back and the names to send them under, or everything
    /// the key holds when nobody named any.
    ///
    /// The first half is where the value is read from and the second is what it
    /// comes back as, which are the same bytes unless the client asked for a
    /// field the schema renamed: `RETURN 1 bb` over `body AS bb` reads `body`
    /// off the key and answers it under `bb`.
    ret: Option<Returns<'a>>,
    /// The text fields the query is narrowed to, or every field.
    infields: Option<Vec<&'a [u8]>>,
    /// The only keys that may answer, when the client listed them.
    inkeys: Option<Vec<&'a [u8]>>,
    /// The numeric ranges hung on the query from outside it.
    filters: Vec<(&'a [u8], f64, f64)>,
    /// The circles hung on the query from outside it, which is the same idea
    /// written for a geo field. Owned rather than borrowed like the ranges
    /// above, because a circle carries the unit as the client spelled it and
    /// the query tree wants that on the node.
    circles: Vec<Circle>,
    scorer: Scorer,
    /// What `HAMMING` compares each document's payload against.
    payload: Option<&'a [u8]>,
    /// Whether an `EXPLAINSCORE` was read, which has nothing to explain unless
    /// something asked for the scores as well.
    explaining: bool,
    /// The room and the order the whole query is read under, which is the same
    /// thing an attribute clause hangs on one group of it.
    slop: Option<i64>,
    inorder: bool,
    /// The field the client wants the answer in the order of and which way
    /// round, as it was written, before the schema has been looked at.
    sort: Option<(&'a [u8], bool)>,
    /// The same once the schema has been looked at, which is what the sort runs
    /// off. Filled in when the whole argument list has read cleanly, because a
    /// real server looks the property up last: `SORTBY 2 @n ASC` complains
    /// about the `@n` at position 3 and not about a property called `2`.
    sorting: Option<Sorting>,
    /// Whether the value the sort compared goes on every row.
    sortkeys: bool,
    /// What a `SUMMARIZE` asked for and which fields it touches, where an empty
    /// list means every field the row carries.
    summarize: Option<(Vec<&'a [u8]>, Trim)>,
    /// The same for a `HIGHLIGHT`.
    highlight: Option<(Vec<&'a [u8]>, Wrap)>,
    /// The names a `FIELDS` list moved to the front of every row, in the order
    /// the clauses named them. Two clauses each with a list put both lists in
    /// front, the one written first coming first.
    upfront: Vec<&'a [u8]>,
    /// The text fields of the schema, copied out because the marking happens
    /// once the registry has been let go of. A field that is not one of these
    /// is cut down but never marked.
    texts: Vec<Box<[u8]>>,
    /// The stop words this index was built with, copied out for the same
    /// reason. A word on the list is free in the arithmetic that decides how
    /// wide a fragment comes back, so the summary of an English sentence is not
    /// the summary the same sentence would get under `STOPWORDS 0`.
    stops: Option<Vec<Box<[u8]>>>,
    /// What the query was looking for, flattened out of the tree before the tree
    /// went away with the registry lock. Empty unless one of the two clauses is
    /// there to use it.
    wanted: Wanted,
    /// The distances the query asks to see on every row, outermost clause
    /// first, taken off the tree in the same place and for the same reason.
    ///
    /// A vector clause puts its distance on the row under `__v_score` or under
    /// the name the client gave it, and a `SORTBY` and a `RETURN` can both name
    /// it. Empty for a query with no vector clause in it, which is why nothing
    /// else here has to ask whether there was one.
    distance: Vec<Yield>,
    /// The distance the rows themselves are in the order of, which only an
    /// aggregation sets.
    ///
    /// A nearest neighbour clause hands its documents back nearest first. A
    /// search then sorts that away, because it orders by score and every
    /// document a vector clause answered scores the same, so what a client sees
    /// is document order. An aggregation does no such sort, so the rows keep
    /// the order the clause made and ties fall back to document order.
    nearest: Option<Box<[u8]>>,
}

/// What a `SORTBY` on a search asked for, once the schema has been read.
struct Sorting {
    /// The field, which carries both the name the client sorted by and the name
    /// the key holds the value under. They are the same unless the schema
    /// renamed it, and the value goes into the reply under the first of them.
    field: Field,
    /// Whether the answer runs backwards.
    desc: bool,
    /// Which of the document's sortable values this is, or nothing when the
    /// index keeps no copy of the field and the key has to be read for it.
    slot: Option<usize>,
    /// Whether this is sorting by a distance the query yielded rather than by
    /// anything the key holds, in which case there is no field to read and the
    /// number is worked out where the answer is gathered.
    distance: bool,
}

impl Rows<'_> {
    /// Whether a row carries a field array at all, which `NOCONTENT` and a
    /// `RETURN` of nothing both take away.
    fn loading(&self) -> bool {
        self.content && !self.ret.as_ref().is_some_and(Vec::is_empty)
    }
}

impl Rows<'_> {
    /// What a search puts on each row, where the fields are gone when a
    /// `RETURN` of nothing emptied the list as well as when `NOCONTENT` did.
    fn found(&self) -> Shows {
        Shows {
            fields: self.loading(),
            scores: self.scores,
            payloads: self.payloads,
            sortkeys: self.sortkeys,
            ..Shows::default()
        }
    }
}

impl Asked<'_> {
    /// What an aggregation puts on each row.
    fn rolls(&self) -> Shows {
        Shows {
            fields: self.rows.content,
            scores: self.rows.scores,
            payloads: self.rows.payloads,
            sortkeys: self.pipe.sortkeys,
            addscores: self.pipe.addscores,
        }
    }
}

impl Default for Rows<'_> {
    fn default() -> Rows<'static> {
        Rows {
            content: true,
            scores: false,
            payloads: false,
            offset: 0,
            count: 10,
            ret: None,
            infields: None,
            inkeys: None,
            filters: Vec::new(),
            circles: Vec::new(),
            scorer: Scorer::default_scorer(),
            payload: None,
            explaining: false,
            slop: None,
            inorder: false,
            sort: None,
            sorting: None,
            sortkeys: false,
            summarize: None,
            highlight: None,
            upfront: Vec::new(),
            texts: Vec::new(),
            stops: None,
            wanted: Wanted::default(),
            distance: Vec::new(),
            nearest: None,
        }
    }
}

/// What goes on a row of a reply beside the fields it carries.
///
/// Copied out of the argument list rather than read from it, so that whoever
/// writes a row does not have to hold on to the words it was asked for with.
#[derive(Clone, Copy, Default)]
struct Shows {
    /// Whether the fields or the properties are on the row at all, which
    /// `NOCONTENT` takes away from a search and from an aggregation both.
    fields: bool,
    scores: bool,
    payloads: bool,
    sortkeys: bool,
    /// Whether the score of the document goes on the row as a property, which
    /// is an aggregation's `ADDSCORES` and nothing a search takes.
    addscores: bool,
}

/// What a client asked for beside the query, and where the reading of it stops.
struct Asked<'a> {
    dialect: u8,
    params: Vec<Pair>,
    verbatim: bool,
    stopwords: bool,
    rows: Rows<'a>,
    pipe: Pipe<'a>,
    /// What a `WITHCURSOR` asked for, when one was asked for at all.
    ///
    /// `FT.EXPLAIN` reads the word and everything after it and then throws all
    /// of it away, the same as it does with `WITHSORTKEYS`.
    cursor: Option<Asks>,
}

impl Default for Asked<'_> {
    fn default() -> Asked<'static> {
        Asked {
            dialect: 1,
            params: Vec::new(),
            verbatim: false,
            stopwords: true,
            rows: Rows::default(),
            pipe: Pipe::default(),
            cursor: None,
        }
    }
}

/// Which of the three commands is reading the argument list.
///
/// They share nearly all of it and part company in five places: `FILTER` is a
/// numeric range on one and an expression on the other two, `LIMIT` is capped
/// on one and not on the rest, `SORTBY` names one bare field on one and a
/// counted list on the other two, three of the words a search takes are refused
/// by name on the aggregation, and `FT.EXPLAIN` reads three words it has
/// nothing to do with so that a search pasted in front of it is refused in the
/// same place.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Mode {
    Search,
    Explain,
    Aggregate,
}

/// Whether reading the arguments is also tying every property a step names to
/// a place on the row, and what distances the query yields when it is.
///
/// An aggregation reads its arguments twice. The names a step is allowed to use
/// include the ones the query yields, and the query cannot be read until the
/// words that say how to read it have been, so the first pass takes the words
/// and looks nothing up and the second pass does the looking up with the query
/// already parsed. That is the order a real server does it in and it shows:
/// `FT.AGGREGATE i "foo(" LIMIT x 1` is refused for the `LIMIT` and
/// `FT.AGGREGATE i "foo(" APPLY '@zz' AS x` is refused for the query, so every
/// word is checked before the query is read and every property is looked up
/// after it.
#[derive(Clone, Copy)]
enum Bind<'a> {
    Reading,
    Binding(&'a [Yield]),
}

/// The order the documents that answered come back in.
///
/// A search ranks them. An aggregation with nothing sorting it hands them back
/// in the order the index holds them, and backwards when its scorer had to see
/// the whole answer before any of it could be written.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Order {
    Ranked,
    Forwards,
    Backwards,
}

/// Whether the whole answer has to exist before the first row of it can be
/// written, which is true of one scorer and only when its scores are asked for.
///
/// `BM25STD.NORM` divides every score by the best one in the answer, so a row
/// cannot be finished until the last of them has been seen. A real server shows
/// that in two ways at once: the count at the front of the reply is the real
/// total whatever the window is, and the rows come back in descending document
/// number rather than ascending. Both go away without `ADDSCORES`, because
/// without it nothing on the row needs the score.
fn buffered(asked: &Asked<'_>) -> bool {
    asked.pipe.addscores && asked.rows.scorer.settles()
}

/// The keywords `FT.EXPLAIN` takes and drops, with how many words each carries.
///
/// Both are here because a real server reads them on that command and this one
/// has nothing to do with them, since both are about the rows of an answer it
/// never sends. `EXPLAINSCORE` is not one of them: every command reads that
/// word and every command refuses it without something asking for the scores,
/// so it is read in the ordinary place and checked in the ordinary place.
const IGNORED: &[(&[u8], usize)] = &[(b"WITHSORTKEYS", 0), (b"ADDSCORES", 0)];

/// Reads the arguments after the query.
///
/// The error text carries a position, which is why this hands back bytes rather
/// than a `Fail`: every other error line in this module is three static pieces
/// around a word the client sent, and this one has a number in it.
fn options<'a>(
    args: Args<'a>,
    from: usize,
    mode: Mode,
    index: &Index,
    bind: Bind<'_>,
) -> core::result::Result<Asked<'a>, Vec<u8>> {
    let main = mode == Mode::Search;
    let mut asked = Asked::default();
    if let Bind::Binding(held) = bind {
        asked.pipe.binding = true;
        // The distances go on the row before anything a `LOAD` asked for,
        // which is measured: a query with a nearest neighbour clause in it
        // answers `__v_score` first on every row whether or not the pipeline
        // ever mentions it.
        for want in held {
            asked
                .pipe
                .base
                .push((want.name.clone(), aggregate::Reads::Distance));
        }
    }
    if mode == Mode::Aggregate {
        // A search hands back ten rows when nobody said how many and an
        // aggregation hands back all of them, and the cap a search puts on the
        // width of its window is not here either: `LIMIT 0 1000001` is refused
        // by one command and taken by the other.
        asked.rows.count = usize::MAX;
    }
    let mut at = from;
    while at < args.len() {
        let word = args.get(at);
        // `FT.EXPLAIN` reads the pipeline as well, and reads it the same way,
        // because a real server hands the whole argument list to one parser and
        // then prints the query out of what came back. So a step is checked on
        // that command and thrown away, and a step closes the door on the words
        // about the search there too: `LOAD 1 @t VERBATIM` is refused on both.
        if mode != Mode::Search
            && let Some(next) = step(args, at, &mut asked, index)?
        {
            at = next;
            continue;
        }
        if args::is(word, b"DIALECT") {
            let Some(value) = args.opt(at + 1) else {
                return Err(line(NEED_ARG, b"DIALECT", ""));
            };
            let Some(dialect) = parse_i64(value).filter(|d| (1..=i64::from(NEWEST)).contains(d))
            else {
                return Err(BAD_DIALECT.as_bytes().to_vec());
            };
            asked.dialect = u8::try_from(dialect).unwrap_or(1);
            at += 2;
            continue;
        }
        if args::is(word, b"PARAMS") {
            at = params(args, at, &mut asked)?;
            continue;
        }
        // The words about the search itself, which an aggregation stops taking
        // once a step of its pipeline has been read. `LIMIT` and `TIMEOUT` are
        // not among them, which is why they are asked for before this.
        if let Some(next) = plan(args, at, &mut asked, mode)? {
            at = next;
            continue;
        }
        if !asked.pipe.stepped {
            if args::is(word, b"VERBATIM") {
                asked.verbatim = true;
                at += 1;
                continue;
            }
            if args::is(word, b"NOSTOPWORDS") {
                asked.stopwords = false;
                at += 1;
                continue;
            }
            if mode == Mode::Aggregate
                && let Some(next) = extra(args, at, &mut asked)?
            {
                at = next;
                continue;
            }
            if let Some(next) = row(args, at, &mut asked.rows, main, index)? {
                at = next;
                continue;
            }
            if mode == Mode::Explain
                && let Some((_, takes)) = IGNORED.iter().find(|(k, _)| args::is(word, k))
            {
                if args.opt(at + takes).is_none() {
                    return Err(line(BAD_ARGS, word, NOT_THERE));
                }
                at += takes + 1;
                continue;
            }
        }
        return Err(unknown(word, at - from + 1));
    }
    // Kept until here because a `LOAD` that ran out on its `AS` is only refused
    // when nothing else was wrong with the list.
    if asked.pipe.pending {
        return Err(NEED_LOAD_NAME.as_bytes().to_vec());
    }
    // Both of these are asked at the end because a real server asks them at the
    // end: an unknown argument anywhere in the list is answered before either,
    // and `WITHSCORES` counts for an `EXPLAINSCORE` that came before it.
    if asked.rows.explaining && !asked.rows.scores {
        return Err(SCORE_ALONE.as_bytes().to_vec());
    }
    aggregate::most(&mut asked);
    Ok(asked)
}

/// Looks the `SORTBY` field up, once the query has been parsed and the tree has
/// said which distances the rows will carry.
///
/// Left until after the parse because a real server leaves it until after the
/// parse: `FT.SEARCH i "foo(" SORTBY zz` answers about the query and not about
/// `zz`, and a sort by `__v_score` is a sort by a property that only exists
/// because the query put it there.
fn settle<'a>(
    asked: &mut Asked<'a>,
    index: &Index,
    node: &Node,
) -> core::result::Result<(), Vec<u8>> {
    asked.rows.distance = query::yields(node);
    let Some((field, desc)) = asked.rows.sort else {
        return Ok(());
    };
    if asked.rows.distance.iter().any(|held| *held.name == *field) {
        asked.rows.sorting = Some(Sorting {
            field: Field::new(field, Kind::Numeric),
            desc,
            slot: None,
            distance: true,
        });
        return Ok(());
    }
    let Some(known) = index.field(field) else {
        return Err(line(SORT_PROP, field, SORT_PROP_END));
    };
    asked.rows.sorting = Some(Sorting {
        field: known.clone(),
        desc,
        slot: index.slot(field),
        distance: false,
    });
    Ok(())
}

/// `LIMIT` and `TIMEOUT`, which every one of the three commands takes wherever
/// they appear.
fn plan<'a>(
    args: Args<'a>,
    at: usize,
    asked: &mut Asked<'a>,
    mode: Mode,
) -> core::result::Result<Option<usize>, Vec<u8>> {
    let rows = &mut asked.rows;
    let word = args.get(at);
    if args::is(word, b"LIMIT") {
        let (Some(offset), Some(count)) = (args.opt(at + 1), args.opt(at + 2)) else {
            return Err(LIMIT_TWO.as_bytes().to_vec());
        };
        let (Some(offset), Some(count)) = (counted(offset), counted(count)) else {
            return Err(LIMIT_NUMBERS.as_bytes().to_vec());
        };
        // The cap is on the width of the window and not on where it starts, so
        // `LIMIT 999999 1000000` is a query a real server takes and
        // `LIMIT 0 1000001` is one it refuses. It is a search's cap alone: the
        // rows an aggregation hands back are not a window on a scored answer,
        // they are what its pipeline made, and `FT.EXPLAIN` never sends a row
        // at all. Both take `LIMIT 0 1000001` and both still refuse
        // `LIMIT 1 0`.
        if count > MOST && mode == Mode::Search {
            return Err(LIMIT_OVER.as_bytes().to_vec());
        }
        // A window of nothing is a client asking for the total on its own, and
        // asking for the total starting from somewhere is a contradiction the
        // three commands all refuse in the same words.
        if count == 0 && offset != 0 {
            return Err(LIMIT_START.as_bytes().to_vec());
        }
        rows.offset = usize::try_from(offset).unwrap_or(0);
        rows.count = usize::try_from(count).unwrap_or(0);
        // An aggregation puts the window where the client wrote it rather than
        // over the answer, so it goes into the pipeline as well. What stays
        // here is what the count at the front of the reply is worked out from.
        if mode == Mode::Aggregate {
            let (offset, count) = (rows.offset, rows.count);
            windows(asked, offset, count);
        }
        return Ok(Some(at + 3));
    }
    if args::is(word, b"WITHCURSOR") {
        // Not a pipeline step, so it is taken wherever it appears and does not
        // close the door on the words about the search itself:
        // `WITHCURSOR COUNT 2 VERBATIM` is a query and `LOAD 1 @t VERBATIM` is
        // not. A second `WITHCURSOR` keeps what the first one set rather than
        // going back to the defaults, which is measured.
        let mut asks = asked.cursor.unwrap_or_default();
        let mut next = at + 1;
        while next < args.len() {
            let word = args.get(next);
            let count = args::is(word, b"COUNT");
            if !count && !args::is(word, b"MAXIDLE") {
                break;
            }
            // The name in the line is the keyword and not the word the client
            // typed, so a lower case `count 0` is refused about `COUNT`.
            let name: &[u8] = match count {
                true => b"COUNT",
                false => b"MAXIDLE",
            };
            let Some(value) = args.opt(next + 1) else {
                return Err(line(BAD_ARGS, name, NOT_THERE));
            };
            let Some(number) = parse_i64(value) else {
                return Err(line(BAD_ARGS, name, NOT_A_NUMBER));
            };
            if !(1..=cursor::MOST).contains(&number) {
                return Err(line(BAD_ARGS, name, OUT_OF_RANGE));
            }
            match count {
                true => asks.count = usize::try_from(number).unwrap_or(usize::MAX),
                false => asks.idle = u64::try_from(number).unwrap_or(u64::MAX),
            }
            next += 2;
        }
        asked.cursor = Some(asks);
        return Ok(Some(next));
    }
    if args::is(word, b"TIMEOUT") {
        let Some(value) = args.opt(at + 1) else {
            return Err(TIMEOUT_ARG.as_bytes().to_vec());
        };
        if counted(value).is_none() {
            return Err(TIMEOUT_NUMBER.as_bytes().to_vec());
        }
        // Nothing here runs long enough to time out, and a deadline that is
        // never reached is the same as no deadline, so the number is checked
        // and dropped.
        return Ok(Some(at + 2));
    }
    Ok(None)
}

/// A step of the pipeline, or nothing when this is not one of those.
///
/// `LOAD`, `GROUPBY`, `APPLY` and `FILTER` are the four read here. `SORTBY` and
/// `LIMIT` are steps too and are read beside the words about the search, since
/// a search takes both of them under the same names.
///
/// A step is read wherever it appears, and reading one closes the door on the
/// words about the search itself. That is what makes `LOAD 1 @t VERBATIM` a
/// refusal and `LIMIT 0 1 VERBATIM` a query.
fn step<'a>(
    args: Args<'a>,
    at: usize,
    asked: &mut Asked<'a>,
    index: &Index,
) -> core::result::Result<Option<usize>, Vec<u8>> {
    if args::is(args.get(at), b"GROUPBY") {
        return group(args, at, asked, index).map(Some);
    }
    if args::is(args.get(at), b"APPLY") {
        return apply(args, at, asked, index).map(Some);
    }
    if args::is(args.get(at), b"FILTER") {
        return keeps(args, at, asked, index).map(Some);
    }
    if args::is(args.get(at), b"SORTBY") {
        return sorts(args, at, asked, index).map(Some);
    }
    if !args::is(args.get(at), b"LOAD") {
        return Ok(None);
    }
    // A `LOAD` reads fields off a key, and once a group step has run there is no
    // key under the row any more. A real server names that rather than letting
    // it read and answer nothing.
    if asked.pipe.stage.is_some() {
        return Err(LOAD_LATE.as_bytes().to_vec());
    }
    let Some(count) = args.opt(at + 1) else {
        return Err(LOAD_SHORT.as_bytes().to_vec());
    };
    // A `LOAD` is a step of the pipeline whatever it names, and a `LOAD 0` is a
    // step that names nothing, so it shuts the door on the words about the
    // search without becoming a loader: `LOAD 0 VERBATIM` is refused and the
    // count at the front of the reply is the one a query with no `LOAD` gets.
    asked.pipe.stepped = true;
    if count == b"*" {
        asked.pipe.all = true;
        asked.pipe.loader = true;
        asked.pipe.loads = true;
        return Ok(Some(at + 2));
    }
    let Some(count) = parse_i64(count) else {
        return Err(LOAD_COUNT.as_bytes().to_vec());
    };
    let Ok(count) = usize::try_from(count) else {
        return Err(LOAD_BOUNDS.as_bytes().to_vec());
    };
    asked.pipe.loader |= count > 0;
    let mut at = at + 2;
    let end = at + count;
    while at < end {
        let Some(path) = args.opt(at) else {
            return Err(LOAD_SHORT.as_bytes().to_vec());
        };
        at += 1;
        // The count is a word count and not a field count, so the `AS` and the
        // name after it are two of the words it pays for. A count that stops on
        // the `AS` is a complaint held back until the rest of the list has read
        // cleanly, because the word after it is read as an argument of its own
        // and may well be worth an error of its own.
        // The `@` is part of how a path is written and not part of the name the
        // property comes back under, so `LOAD 1 @t` and `LOAD 1 t` answer the
        // same `t`.
        let field = path.strip_prefix(b"@").unwrap_or(path);
        let name = match at < end && args::is(args.get(at), b"AS") {
            false => field,
            true => {
                at += 1;
                match args.opt(at).filter(|_| at < end) {
                    None => {
                        asked.pipe.pending = true;
                        break;
                    }
                    Some(name) => {
                        at += 1;
                        name
                    }
                }
            }
        };
        asked.pipe.load.push((field, name));
        // A sortable field is held beside the document number, so naming one is
        // not on its own a reason to open the key.
        asked.pipe.loads |= !index.field(field).is_some_and(|held| held.sortable);
        // The same pair again for the pipeline, which reads a row by position
        // rather than by name. A name loaded twice is answered once and located
        // once, so the second copy is dropped here rather than later.
        if !asked.pipe.base.iter().any(|(held, _)| **held == *name) {
            asked
                .pipe
                .base
                .push((name.into(), Reads::Field(field.into(), holds(index, field))));
        }
    }
    Ok(Some(at))
}

/// One of the words `FT.AGGREGATE` alone reads, or nothing when this is not one
/// of those.
fn extra<'a>(
    args: Args<'a>,
    at: usize,
    asked: &mut Asked<'a>,
) -> core::result::Result<Option<usize>, Vec<u8>> {
    let word = args.get(at);
    if args::is(word, b"ADDSCORES") {
        asked.pipe.addscores = true;
        return Ok(Some(at + 1));
    }
    if args::is(word, b"WITHSORTKEYS") {
        asked.pipe.sortkeys = true;
        return Ok(Some(at + 1));
    }
    // Three words a search takes that an aggregation names in its refusal
    // rather than letting them fall through to the unknown argument line.
    for name in [b"RETURN".as_slice(), b"SUMMARIZE", b"HIGHLIGHT"] {
        if args::is(word, name) {
            return Err(line("SEARCH_PARSE_ARGS ", name, NOT_HERE));
        }
    }
    Ok(None)
}

/// One argument about the rows, or nothing when this is not one of those.
///
/// Everything in here is read by both commands and acted on by one of them, so
/// a search that a client pastes in front of `FT.EXPLAIN` is refused in the
/// same place for the same reason as the search itself.
fn row<'a>(
    args: Args<'a>,
    at: usize,
    rows: &mut Rows<'a>,
    main: bool,
    index: &Index,
) -> core::result::Result<Option<usize>, Vec<u8>> {
    let word = args.get(at);
    if args::is(word, b"NOCONTENT") {
        rows.content = false;
        return Ok(Some(at + 1));
    }
    if args::is(word, b"WITHSCORES") {
        rows.scores = true;
        return Ok(Some(at + 1));
    }
    if args::is(word, b"WITHPAYLOADS") {
        rows.payloads = true;
        return Ok(Some(at + 1));
    }
    // Read here on all three commands, because all three read it. Two of them
    // have no score to explain and answer so once the whole list has been read,
    // rather than refusing the word where it stands.
    if args::is(word, b"EXPLAINSCORE") {
        rows.explaining = true;
        return Ok(Some(at + 1));
    }
    if args::is(word, b"INORDER") {
        rows.inorder = true;
        return Ok(Some(at + 1));
    }
    if args::is(word, b"SLOP") {
        let Some(value) = args.opt(at + 1) else {
            return Err(line(BAD_ARGS, word, NOT_THERE));
        };
        let Some(slop) = parse_i64(value) else {
            return Err(line(BAD_ARGS, word, NOT_A_NUMBER));
        };
        rows.slop = Some(slop);
        return Ok(Some(at + 2));
    }
    if args::is(word, b"SCORER") {
        let Some(name) = args.opt(at + 1) else {
            return Err(line(BAD_ARGS, word, NOT_THERE));
        };
        let Some(scorer) = Scorer::named(name) else {
            return Err(line(NO_SCORER, name, ""));
        };
        rows.scorer = scorer;
        return Ok(Some(at + 2));
    }
    if args::is(word, b"LANGUAGE") {
        let Some(name) = args.opt(at + 1) else {
            return Err(line(BAD_ARGS, word, NOT_THERE));
        };
        if !LANGUAGES.iter().any(|known| args::is(name, known)) {
            return Err(NO_LANGUAGE.as_bytes().to_vec());
        }
        // The index was built in one language and the words in it were stemmed
        // in that language, so a query stemmed in another would be asking the
        // index for words it does not hold. The name is checked and dropped.
        return Ok(Some(at + 2));
    }
    if args::is(word, b"EXPANDER") || args::is(word, b"PAYLOAD") {
        let Some(value) = args.opt(at + 1) else {
            return Err(line(BAD_ARGS, word, NOT_THERE));
        };
        if args::is(word, b"PAYLOAD") {
            rows.payload = Some(value);
        }
        // An expander is a module a real server loads and there is nowhere to
        // load one from here, so any name at all is taken and nothing is done
        // with it, which is what a real server does with a name it does not
        // know as well.
        return Ok(Some(at + 2));
    }
    if args::is(word, b"RETURN") {
        return returned(args, at, rows, index).map(Some);
    }
    if args::is(word, b"SUMMARIZE") || args::is(word, b"HIGHLIGHT") {
        return marking(args, at, rows, index).map(Some);
    }
    if main && args::is(word, b"SORTBY") {
        return sorting(args, at, rows).map(Some);
    }
    if main && args::is(word, b"WITHSORTKEYS") {
        rows.sortkeys = true;
        return Ok(Some(at + 1));
    }
    if args::is(word, b"INFIELDS") {
        let (names, next) = names(args, at)?;
        rows.infields = Some(names);
        return Ok(Some(next));
    }
    if args::is(word, b"INKEYS") {
        let (names, next) = names(args, at)?;
        rows.inkeys = Some(names);
        return Ok(Some(next));
    }
    if main && args::is(word, b"FILTER") {
        return filter(args, at, rows, index).map(Some);
    }
    if main && args::is(word, b"GEOFILTER") {
        return geofilter(args, at, rows).map(Some);
    }
    Ok(None)
}

/// `SORTBY field [ASC|DESC]` on a search, which is not the shape an
/// aggregation's sort step takes: one bare field name, no count in front of it
/// and no `@` on it. `SORTBY @n` is refused for a property called `@n`.
///
/// The direction is optional and either word will do, in any case. Anything
/// else after the field belongs to whatever comes next in the list, so
/// `SORTBY n DESC DESC` is a sort followed by an unknown argument. `MAX` is the
/// exception: a sort step takes one and a search has nowhere to put it, and the
/// word is recognised in both places after the field so that a client hears
/// what is wrong rather than that the word is unknown.
///
/// A second `SORTBY` is refused before its field is read, which is measured:
/// `SORTBY n SORTBY` with nothing after it answers about the second sort and
/// not about the argument it is missing.
fn sorting<'a>(
    args: Args<'a>,
    at: usize,
    rows: &mut Rows<'a>,
) -> core::result::Result<usize, Vec<u8>> {
    if rows.sort.is_some() {
        return Err(SORT_ONCE.as_bytes().to_vec());
    }
    let Some(field) = args.opt(at + 1) else {
        return Err(SORT_BARE.as_bytes().to_vec());
    };
    let mut next = at + 2;
    let mut desc = false;
    if let Some(word) = args.opt(next)
        && (args::is(word, b"ASC") || args::is(word, b"DESC"))
    {
        desc = args::is(word, b"DESC");
        next += 1;
    }
    if args.opt(next).is_some_and(|word| args::is(word, b"MAX")) {
        return Err(SORT_MAX.as_bytes().to_vec());
    }
    rows.sort = Some((field, desc));
    Ok(next)
}

/// A number that is a whole number and not below zero, which is what `LIMIT`
/// and `TIMEOUT` want.
fn counted(value: &[u8]) -> Option<i64> {
    parse_i64(value).filter(|n| *n >= 0)
}

/// `RETURN n field [AS name] ...`, where `n` counts words and not fields.
///
/// That is measured and it is the one thing about this argument that surprises
/// everybody: `RETURN 2 t NOCONTENT` reads `NOCONTENT` as the name of a second
/// field, so the rows come back with their content after all. `RETURN 0` names
/// no fields, which reads on the wire like `NOCONTENT` and is not the same
/// thing, because a `RETURN` after it puts the fields back and a `RETURN` after
/// a `NOCONTENT` does not.
///
/// The count covers the `AS` and the name after it as well as the field, and a
/// count that reaches the `AS` without reaching the name is its own error
/// rather than a field called `AS`. A count that stops before the `AS` is not:
/// `RETURN 1 AS` asks for a field called `AS`, which no key holds.
///
/// A name the schema knows is read off the key under the identifier that schema
/// gave it, so `RETURN 1 bb` over `body AS bb` answers what the key holds under
/// `body`. A name the schema does not know is read off the key as it stands.
fn returned<'a>(
    args: Args<'a>,
    at: usize,
    rows: &mut Rows<'a>,
    index: &Index,
) -> core::result::Result<usize, Vec<u8>> {
    let Some(count) = args.opt(at + 1) else {
        return Err(line(BAD_ARGS, b"RETURN", NOT_THERE));
    };
    let Some(count) = parse_i64(count).filter(|n| *n >= 0) else {
        return Err(line(BAD_ARGS, b"RETURN", NOT_A_NUMBER));
    };
    let count = usize::try_from(count).unwrap_or(0);
    let mut want = Vec::new();
    let mut step = 0;
    while step < count {
        let Some(field) = args.opt(at + 2 + step) else {
            return Err(line(BAD_ARGS, b"RETURN", NOT_THERE));
        };
        step += 1;
        if step < count && args.opt(at + 2 + step).is_some_and(|w| args::is(w, b"AS")) {
            // The count stopped on the `AS`, so whatever name follows belongs
            // to what comes after `RETURN` rather than to this field, and the
            // rename has nothing to rename to. That is asked before the name is
            // looked for, because `RETURN 2 t AS` with nothing after it at all
            // answers this line and not the one about a missing argument.
            if step + 1 >= count {
                return Err(NEED_NAME.as_bytes().to_vec());
            }
            let Some(name) = args.opt(at + 3 + step) else {
                return Err(line(BAD_ARGS, b"RETURN", NOT_THERE));
            };
            want.push((reads(index, field), name));
            step += 2;
            continue;
        }
        want.push((reads(index, field), field));
    }
    // The last list wins rather than the lists adding up, so
    // `RETURN 1 t RETURN 1 b` answers `b` on its own.
    rows.ret = Some(want);
    Ok(at + 2 + count)
}

/// `SUMMARIZE [FIELDS n name...] [FRAGS n] [LEN n] [SEPARATOR s]`, or
/// `HIGHLIGHT [FIELDS n name...] [TAGS open close]`.
///
/// The two are one function because everything about reading them is shared
/// except which words they take after the `FIELDS` list. Writing one of those
/// words under the other clause is not an error about the clause, it is the
/// clause ending and an unknown argument beginning, so `SUMMARIZE TAGS a b`
/// complains about `TAGS` at its own position.
///
/// A `FIELDS` count of nothing means every field, which is also what leaving
/// the list out means. A name the schema does not hold is refused, and the `@`
/// a query would write in front of a field counts as part of the name, so `@a`
/// is not a property either.
///
/// Every other way of getting it wrong answers one line: a missing value, a
/// value that is not a number, a negative number, and a count that promises
/// more names than the list has left. Both clauses may be written more than
/// once and the last one wins, which is why the settings start again from the
/// defaults each time round rather than from what the last one left.
fn marking<'a>(
    args: Args<'a>,
    at: usize,
    rows: &mut Rows<'a>,
    index: &Index,
) -> core::result::Result<usize, Vec<u8>> {
    let word = args.get(at);
    let summary = args::is(word, b"SUMMARIZE");
    let bad = || line(BAD_ARGS, word, "");
    let mut at = at + 1;
    let mut fields: Vec<&'a [u8]> = Vec::new();
    let mut trim = Trim::default();
    let mut wrap = Wrap::default();
    while let Some(next) = args.opt(at) {
        if args::is(next, b"FIELDS") {
            let Some(count) = args.opt(at + 1).and_then(parse_i64).filter(|n| *n >= 0) else {
                return Err(bad());
            };
            let count = usize::try_from(count).unwrap_or(0);
            for step in 0..count {
                let Some(name) = args.opt(at + 2 + step) else {
                    return Err(bad());
                };
                if !index.schema.iter().any(|f| *f.attribute == *name) {
                    return Err(line(NO_PROPERTY, name, QUOTE_END));
                }
                fields.push(name);
            }
            at += 2 + count;
            continue;
        }
        if summary && (args::is(next, b"FRAGS") || args::is(next, b"LEN")) {
            let Some(value) = args.opt(at + 1).and_then(parse_i64).filter(|n| *n >= 0) else {
                return Err(bad());
            };
            let value = usize::try_from(value).unwrap_or(0);
            match args::is(next, b"FRAGS") {
                true => trim.frags = value,
                false => trim.len = value,
            }
            at += 2;
            continue;
        }
        if summary && args::is(next, b"SEPARATOR") {
            let Some(value) = args.opt(at + 1) else {
                return Err(bad());
            };
            trim.separator = value.into();
            at += 2;
            continue;
        }
        if !summary && args::is(next, b"TAGS") {
            let (Some(open), Some(close)) = (args.opt(at + 1), args.opt(at + 2)) else {
                return Err(bad());
            };
            wrap.open = open.into();
            wrap.close = close.into();
            at += 3;
            continue;
        }
        break;
    }
    // The names go in front of every row, in the order the clauses named them,
    // and a name both clauses named only goes in front once.
    for name in &fields {
        if !rows.upfront.contains(name) {
            rows.upfront.push(name);
        }
    }
    if rows.texts.is_empty() {
        rows.texts = index
            .schema
            .iter()
            .filter(|field| matches!(field.kind, Kind::Text(_)))
            .map(|field| field.attribute.clone())
            .collect();
        rows.stops = index.definition.stopwords.clone();
    }
    match summary {
        true => rows.summarize = Some((fields, trim)),
        false => rows.highlight = Some((fields, wrap)),
    }
    Ok(at)
}

/// Where a field a client named by hand is read from on the key.
///
/// The identifier when the schema knows the name, and the name itself when it
/// does not, since a client may ask for a field nobody indexed.
fn reads(index: &Index, name: &[u8]) -> Box<[u8]> {
    index
        .field(name)
        .map_or_else(|| name.into(), |field| field.identifier.clone())
}

/// `INFIELDS n name ...` and `INKEYS n name ...`, which are the same shape.
fn names<'a>(args: Args<'a>, at: usize) -> core::result::Result<(Vec<&'a [u8]>, usize), Vec<u8>> {
    let word = args.get(at);
    let Some(count) = args.opt(at + 1) else {
        return Err(line(BAD_ARGS, word, NOT_THERE));
    };
    let Some(count) = parse_i64(count).filter(|n| *n >= 0) else {
        return Err(line(BAD_ARGS, word, NOT_A_NUMBER));
    };
    let count = usize::try_from(count).unwrap_or(0);
    let mut out = Vec::with_capacity(count);
    for step in 0..count {
        let Some(name) = args.opt(at + 2 + step) else {
            return Err(line(BAD_ARGS, word, NOT_THERE));
        };
        out.push(name);
    }
    Ok((out, at + 2 + count))
}

/// `FILTER field min max`, which is a numeric range written outside the query.
///
/// A field that is not there or is not a number is not an error and answers
/// nothing, which falls out of asking the index for a numeric field it does not
/// have. The two ends are read as numbers whatever the field is, though, so
/// `FILTER nope x 1` is refused for `x`, and only a field the schema really
/// does hold as a number is checked for being the wrong way round:
/// `FILTER n 2 1` is refused and `FILTER nope 2 1` and `FILTER g 2 1` answer
/// nothing.
fn filter<'a>(
    args: Args<'a>,
    at: usize,
    rows: &mut Rows<'a>,
    index: &Index,
) -> core::result::Result<usize, Vec<u8>> {
    let (Some(field), Some(min), Some(max)) =
        (args.opt(at + 1), args.opt(at + 2), args.opt(at + 3))
    else {
        return Err(FILTER_THREE.as_bytes().to_vec());
    };
    let Some(low) = ends(min) else {
        return Err(line(LOW_RANGE, min, ""));
    };
    let Some(high) = ends(max) else {
        return Err(line(HIGH_RANGE, max, ""));
    };
    if low > high && numeric(index, field) {
        return Err(backwards(field, low, high));
    }
    rows.filters.push((field, low, high));
    Ok(at + 4)
}

/// `GEOFILTER field lon lat radius unit`, which is a circle written outside the
/// query.
///
/// The field is not looked at here at all. A name the schema has never heard of
/// and a name it holds as something other than a geo field both answer nothing
/// rather than being refused, which is the same thing a `FILTER` on a field
/// that is not a number does, and it falls out of asking the index for points
/// under a name that has none.
///
/// The five things that are refused are refused in this order, which is
/// measured: too few arguments, then a number that will not parse, then a unit
/// that is not one of the four, then a centre off the projection, then a radius
/// with no size to it.
fn geofilter<'a>(
    args: Args<'a>,
    at: usize,
    rows: &mut Rows<'a>,
) -> core::result::Result<usize, Vec<u8>> {
    let (Some(field), Some(lon), Some(lat), Some(radius), Some(unit)) = (
        args.opt(at + 1),
        args.opt(at + 2),
        args.opt(at + 3),
        args.opt(at + 4),
        args.opt(at + 5),
    ) else {
        return Err(GEO_FIVE.as_bytes().to_vec());
    };
    let mut ends = [0.0; 3];
    // The three names go on the wire inside angle brackets, which nothing else
    // in this family does and is what a real server sends.
    for (slot, (raw, what)) in ends.iter_mut().zip([
        (lon, &b"<lon>"[..]),
        (lat, &b"<lat>"[..]),
        (radius, &b"<radius>"[..]),
    ]) {
        let Some(value) = coord(raw) else {
            return Err(line(BAD_ARGS, what, NOT_A_NUMBER));
        };
        *slot = value;
    }
    if geo::Unit::parse(unit).is_none() {
        return Err(line(GEO_UNIT, unit, ""));
    }
    // A NaN goes through both of these, since it is neither outside the limits
    // nor at or below zero, and then answers nothing because it is inside no
    // circle. That is what a real server does with one.
    if !ends[0].is_nan() && !ends[1].is_nan() && !geo::in_range(ends[0], ends[1]) {
        return Err(line(SYNTAX, BAD_POINT.as_bytes(), ""));
    }
    if ends[2] <= 0.0 {
        return Err(line(SYNTAX, BAD_RADIUS.as_bytes(), ""));
    }
    rows.circles.push(Circle {
        field: field.into(),
        lon: ends[0],
        lat: ends[1],
        radius: ends[2],
        unit: unit.into(),
    });
    Ok(at + 6)
}

/// One of the three numbers a `GEOFILTER` is written with.
///
/// A NaN is read and an infinity is not, which is the other way round from
/// every other number on this command and is measured: `GEOFILTER loc nan 0 1
/// km` answers no rows and `GEOFILTER loc inf 0 1 km` is refused. Space in
/// front is allowed and space behind is not, which is also measured.
fn coord(raw: &[u8]) -> Option<f64> {
    let text = core::str::from_utf8(raw).ok()?;
    let value: f64 = text.trim_ascii_start().parse().ok()?;
    (!value.is_infinite()).then_some(value)
}

/// Whether the schema holds this name as a number, which is the one field a
/// `FILTER` can be the wrong way round on.
fn numeric(index: &Index, field: &[u8]) -> bool {
    index
        .schema
        .iter()
        .any(|f| *f.attribute == *field && matches!(f.kind, Kind::Numeric))
}

/// One end of a `FILTER`, which takes the two infinities by name as well as a
/// number.
fn ends(value: &[u8]) -> Option<f64> {
    match value {
        b"+inf" | b"inf" | b"INF" | b"+INF" => Some(f64::INFINITY),
        b"-inf" | b"-INF" => Some(f64::NEG_INFINITY),
        _ => parse_f64(value),
    }
}

/// `Invalid numeric range (min > max): @n:[2.000000 1.000000]`, which prints
/// both ends to six places however they were written.
fn backwards(field: &[u8], low: f64, high: f64) -> Vec<u8> {
    let mut out = BACKWARDS.as_bytes().to_vec();
    out.extend_from_slice(field);
    out.extend_from_slice(b":[");
    out.extend_from_slice(format!("{low:.6} {high:.6}").as_bytes());
    out.push(b']');
    out
}

/// `PARAMS n name value ...`, which is where a `$name` in a query comes from.
fn params(
    args: Args<'_>,
    at: usize,
    asked: &mut Asked<'_>,
) -> core::result::Result<usize, Vec<u8>> {
    let Some(count) = args.opt(at + 1) else {
        return Err(line(BAD_ARGS, b"PARAMS", NOT_THERE));
    };
    let Some(count) = parse_i64(count) else {
        return Err(line(BAD_ARGS, b"PARAMS", NOT_A_NUMBER));
    };
    let Ok(count) = usize::try_from(count) else {
        return Err(line(BAD_ARGS, b"PARAMS", OUT_OF_RANGE));
    };
    // The words are counted before their shape is looked at, which is the way
    // round a real server does it and is worth keeping because the two errors
    // are different. `PARAMS 3 a b` reaches past the end of the command and
    // says an argument was expected, while `PARAMS 3 a b c` has all three and
    // only then gets told the count has to be even.
    if at + 2 + count > args.len() {
        return Err(line(BAD_ARGS, b"PARAMS", NOT_THERE));
    }
    if count == 0 || count % 2 != 0 {
        return Err(ODD_PARAMS.as_bytes().to_vec());
    }
    for step in 0..count / 2 {
        let name = args.opt(at + 2 + step * 2);
        let value = args.opt(at + 3 + step * 2);
        let (Some(name), Some(value)) = (name, value) else {
            return Err(line(BAD_ARGS, b"PARAMS", NOT_THERE));
        };
        asked.params.push((name.into(), value.into()));
    }
    Ok(at + 2 + count)
}

/// An error line built from a head, a word the client sent and a tail.
fn line(head: &str, word: &[u8], tail: &str) -> Vec<u8> {
    let mut out = head.as_bytes().to_vec();
    out.extend_from_slice(word);
    out.extend_from_slice(tail.as_bytes());
    out
}

/// The line for an argument nobody knows, which counts from the query.
fn unknown(word: &[u8], position: usize) -> Vec<u8> {
    let mut out = UNKNOWN.as_bytes().to_vec();
    out.extend_from_slice(word);
    out.extend_from_slice(NOT_MAIN.as_bytes());
    out.extend_from_slice(position.to_string().as_bytes());
    out.extend_from_slice(NOT_MAIN_END.as_bytes());
    out
}

/// The line a refused query answers with.
fn refused(bad: &Bad) -> Vec<u8> {
    match bad {
        Bad::Syntax { at, near } => spot("SEARCH_SYNTAX Syntax error at offset ", *at, near),
        Bad::Unknown { at, near } => named(
            "SEARCH_SYNTAX Unknown field at offset ",
            *at,
            near.as_deref(),
        ),
        Bad::Wrong { kind, at, near } => {
            let head = format!("SEARCH_SYNTAX Expected a {kind} field at offset ");
            named(&head, *at, near.as_deref())
        }
        Bad::Attribute(name) => line("SEARCH_OPTION_INVALID Invalid attribute ", name, ""),
        Bad::Blob { got, want } => format!(
            "SEARCH_QUERY_BAD Error parsing vector similarity query: query vector blob size ({got}) does not match index's expected size ({want})."
        )
        .into_bytes(),
        Bad::Large => b"SEARCH_QUERY_BAD Error parsing vector similarity query: \
             query KNN K parameter is too large, must not exceed 288230376151711744"
            .to_vec(),
        Bad::Count { name, value } => {
            let mut out = b"SEARCH_NUMERIC_VALUE_INVALID Invalid numeric value (".to_vec();
            out.extend_from_slice(value);
            out.extend_from_slice(b") for parameter `");
            out.extend_from_slice(name);
            out.push(b'`');
            out
        }
        Bad::Radius(radius) => {
            let mut out =
                b"SEARCH_QUERY_BAD Error parsing vector similarity query: negative radius ("
                    .to_vec();
            out.extend_from_slice(radius);
            out.extend_from_slice(b") given in a range query");
            out
        }
        Bad::Value { name, value } => {
            let mut out = b"SEARCH_SYNTAX Invalid value (".to_vec();
            out.extend_from_slice(value);
            out.extend_from_slice(b") for `");
            out.extend_from_slice(name);
            out.push(b'`');
            out
        }
        Bad::Missing(name) => {
            let mut out = b"SEARCH_PARAM_NOT_FOUND Parameter not found `".to_vec();
            out.extend_from_slice(name);
            out.push(b'`');
            out
        }
        Bad::Taken(name) => {
            let mut out = b"SEARCH_INDEX_EXISTS Property `".to_vec();
            out.extend_from_slice(name);
            out.extend_from_slice(b"` already exists in schema");
            out
        }
        // The whole family says which parser was reading at the end of the
        // line, because the same seven sentences come back from a range clause
        // and a nearest neighbour clause and there would otherwise be nothing
        // in them saying it was the vector half of the query that objected.
        Bad::Option(why) => {
            let mut out = why.code().as_bytes().to_vec();
            out.push(b' ');
            out.extend_from_slice(why.words().as_bytes());
            out.extend_from_slice(b" (Error parsing vector similarity parameters)");
            out
        }
        Bad::Twice(first, second) => {
            let mut out =
                b"SEARCH_FIELD_DUP Distance field was specified twice for vector query: ".to_vec();
            out.extend_from_slice(first);
            out.extend_from_slice(b" and ");
            out.extend_from_slice(second);
            out
        }
        Bad::Plain(text) => line("SEARCH_SYNTAX ", text.as_bytes(), ""),
        Bad::Refused(text) => line("SEARCH_QUERY_BAD ", text.as_bytes(), ""),
    }
}

/// The same for the two errors about a field, which leave the `near` off
/// altogether when there is no word to put in it rather than trailing an empty
/// one the way a syntax error does.
fn named(head: &str, at: usize, near: Option<&[u8]>) -> Vec<u8> {
    let Some(near) = near else {
        let mut out = head.as_bytes().to_vec();
        out.extend_from_slice(at.to_string().as_bytes());
        return out;
    };
    spot(head, at, near)
}

/// `... at offset 7 near the`, which is how the parser points at a query.
fn spot(head: &str, at: usize, near: &[u8]) -> Vec<u8> {
    let mut out = head.as_bytes().to_vec();
    out.extend_from_slice(at.to_string().as_bytes());
    out.extend_from_slice(b" near ");
    out.extend_from_slice(near);
    out
}

/// `FT.EXPLAIN index query [options]` and `FT.EXPLAINCLI` beside it.
///
/// The same work with two shapes of reply. `FT.EXPLAIN` sends the printout as
/// one bulk string and `FT.EXPLAINCLI` sends it split on newlines, including
/// the empty piece after the last one, so a client that joins the array back
/// with newlines gets exactly what the other command would have sent.
fn explain<'a>(reg: &mut Registry, args: Args<'a>, out: &mut Out, cli: bool) -> Answer<'a> {
    let name = args.get(1);
    let query = args.get(2);
    // Counted, like every other command that resolves a name, and resolved
    // before the arguments are read rather than after: `FT.EXPLAIN nope q BOGUS`
    // answers about the index and not about the word, and a call that goes on to
    // refuse an argument has still counted as a use.
    let Some(index) = reg.open(name) else {
        return Err(Fail::naming(MISSING, name));
    };
    let asked = match options(args, 3, Mode::Explain, index, Bind::Binding(&[])) {
        Ok(asked) => asked,
        Err(text) => {
            out.error(&text);
            return Ok(());
        }
    };
    let ask = Ask {
        dialect: asked.dialect,
        params: &asked.params,
        verbatim: asked.verbatim,
        stopwords: asked.stopwords,
    };
    let node = match query::parse(query, index, &ask) {
        Ok(node) => node,
        Err(bad) => {
            out.error(&refused(&bad));
            return Ok(());
        }
    };
    let printed = query::explain(&node, index);
    if !cli {
        out.bulk(&printed);
        return Ok(());
    }
    let lines = query::explain::lines(&printed);
    out.array(lines.len());
    for line in lines {
        out.simple(line);
    }
    Ok(())
}

/// One row of the reply, with the fields of its key once they have been read.
type Built<'a> = (&'a Row, Option<Vec<(&'a [u8], &'a [u8])>>);

/// One row of an aggregation, with its properties. There is no `Option` here
/// because a row that carries no property carries an empty list rather than
/// nothing at all.
type Rolled<'a> = (&'a Row, Vec<(&'a [u8], &'a [u8])>);

/// One name and one value a row carries, owned rather than borrowed.
type Named = (Box<[u8]>, Box<[u8]>);

/// The names and values a row carries, owned rather than borrowed. A cursor is
/// written long after the documents it read went away, so it keeps its own copy.
type Pairs = Vec<Named>;

/// One distance a row carries, under the name the row answers it under.
type Away = (Box<[u8]>, f64);

/// A document that answered, what it scored, and the distances it carries.
type Scored<'a> = (walk::Hit<'a>, f64, Vec<Away>);

/// One row of an answer, once the registry has been let go of.
///
/// Nothing in here borrows the index, which is the whole point: the query runs
/// and the scoring happens under the lock, and what comes out is this, so the
/// keys can be read out of the keyspace with the registry free.
#[derive(Clone)]
struct Row {
    key: Box<[u8]>,
    score: f64,
    payload: Option<Box<[u8]>>,
    /// Why the score came out the way it did, when an `EXPLAINSCORE` asked.
    ///
    /// Only the rows in the window carry one, because working it out means
    /// walking what matched a second time and writing a string per branch of
    /// it, and a client only ever reads the rows it was sent.
    note: Option<Note>,
    /// What the sort compared this row on, when a `SORTBY` sorted it and the
    /// document had a value there. Copied out of the index for a field the
    /// index keeps, and read off the key afterwards for a field it does not.
    sort: Option<Sorted>,
    /// The distances the query asked to see, in the order they go on the row.
    ///
    /// Worked out here rather than carried out of the walk, because a query can
    /// ask to see the distance from a clause that ordered nothing, and because
    /// two vector clauses in one query show two distances.
    dists: Vec<Away>,
}

impl Row {
    /// The distance this row carries under one name, when it carries one.
    fn away(&self, name: &[u8]) -> Option<f64> {
        self.dists
            .iter()
            .find(|(held, _)| **held == *name)
            .map(|(_, away)| *away)
    }
}

/// `FT.SEARCH index query [options]`.
///
/// The two halves are described at the top of this file. This is the seam
/// between them, and the lock is held for exactly as long as the first half
/// takes.
pub(super) fn find(server: &Server, db: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    searched(server, db, args, 2, None, out)
}

/// `FT.PROFILE index SEARCH|AGGREGATE [LIMITED] QUERY query [options]`.
///
/// # Errors
///
/// The arity line, when the words run out before the query does.
pub(super) fn profiled(server: &Server, db: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    profile::run(server, db, args, out)
}

/// The same, with the query somewhere other than the third word and with
/// somebody watching.
///
/// `FT.PROFILE` runs a search through here with its own words in front, so what
/// moves is where the query sits and where the options start after it. Nothing
/// else about the search changes, which is the point: a profiled search is the
/// same search.
pub(super) fn searched(
    server: &Server,
    db: usize,
    args: Args<'_>,
    at: usize,
    watch: Option<&mut Watch>,
    out: &mut Out,
) -> Result<()> {
    let mut watch = watch;
    let name = args.get(1);
    let query = args.get(at);
    let clock = Instant::now();
    let mut asked;
    // The name the index is held under, which is not always the name the client
    // used, since an alias reaches the same index. A cursor is kept under the
    // index's own name so that a read through either name finds it.
    let mut canon: Box<[u8]> = Box::default();
    let (total, rows) = {
        let mut reg = server.search.lock();
        // Counted, like every other command that resolves a name, and resolved
        // before the arguments are read: `FT.SEARCH nope q BOGUS` answers about
        // the index and not about the word, and a `FILTER` that names a field
        // cannot be read without the schema in front of it anyway.
        let Some(index) = reg.open(name) else {
            Fail::naming(MISSING, name).write(out);
            return Ok(());
        };
        asked = match options(args, at + 1, Mode::Search, index, Bind::Binding(&[])) {
            Ok(asked) => asked,
            Err(text) => {
                out.error(&text);
                return Ok(());
            }
        };
        let ask = Ask {
            dialect: asked.dialect,
            params: &asked.params,
            verbatim: asked.verbatim,
            stopwords: asked.stopwords,
        };
        let node = match query::parse(query, index, &ask) {
            Ok(node) => node,
            Err(bad) => {
                out.error(&refused(&bad));
                return Ok(());
            }
        };
        // The tree says which distances go on a row and a `SORTBY` may be
        // sorting by one of them, so the field it named is looked up here and
        // not while the arguments were being read.
        if let Err(text) = settle(&mut asked, index, &node) {
            out.error(&text);
            return Ok(());
        }
        if asked.cursor.is_some() {
            canon = index.name.clone();
            // Asked before the query runs, because a real server takes the
            // cursor's place in the table before it makes the reply that goes in
            // it, so an index holding its hundred and twenty eight refuses even
            // a cursor that would close on its first chunk.
            if let Err(fail) = cursor::room(server, &canon, 1) {
                fail.write(out);
                return Ok(());
            }
        }
        // What a `SUMMARIZE` or a `HIGHLIGHT` marks, taken off the parsed
        // query before the tree goes away with the lock.
        if asked.rows.summarize.is_some() || asked.rows.highlight.is_some() {
            asked.rows.wanted = Wanted::of(&node);
        }
        if let Some(watch) = watch.as_deref_mut() {
            watch.parsing = clock.elapsed();
        }
        let shaped = shape(node, index, &asked.rows);
        if let Some(watch) = watch.as_deref_mut() {
            watch.creating = clock.elapsed() - watch.parsing;
        }
        let (total, rows, ran) = gather(
            index,
            shaped,
            &asked.rows,
            Order::Ranked,
            false,
            watch.is_some(),
        );
        if let Some(watch) = watch.as_deref_mut()
            && let Some((ran, spent)) = ran
        {
            watch.ran = Some(ran);
            watch.walking = spent;
        }
        (total, rows)
    };
    if let Some(watch) = watch {
        watch.steps = searching(&asked.rows, total, rows.len());
    }
    write(server, db, total, &rows, &asked, &canon, out);
    Ok(())
}

/// What a search ran the documents that answered through, and how many of them
/// came out of each step.
///
/// Measured against a real server, on which the counts are the number of rows
/// the step handed on rather than the number it was given. The window is what
/// separates the first two from the rest: everything that answered is scored
/// and only what the window kept is sorted, loaded and marked up.
fn searching(rows: &Rows<'_>, total: usize, window: usize) -> Vec<(Vec<u8>, usize)> {
    let mut out = vec![(b"Index".to_vec(), total)];
    // The step that works the distances out, which is there whenever the query
    // yields one and is not there for a range clause nobody named. It sits
    // directly behind the index because every row it touches came off the walk.
    if !rows.distance.is_empty() {
        out.push((b"Metrics Applier".to_vec(), total));
    }
    // A window of nothing is a client asking for the total and nothing else, so
    // there is nothing to score and nothing to sort. Measured: `LIMIT 0 0`
    // answers an index step and a counter and no other step at all.
    if rows.count == 0 {
        out.push((b"Counter".to_vec(), 1));
        return out;
    }
    // A sort by a field does not need a score, and asking for the scores puts
    // the step back whether or not anything is ordered by them.
    if rows.sorting.is_none() || rows.scores {
        out.push((b"Scorer".to_vec(), total));
    }
    out.push((b"Sorter".to_vec(), window));
    if rows.content {
        out.push((b"Loader".to_vec(), window));
        if rows.summarize.is_some() || rows.highlight.is_some() {
            out.push((b"Highlighter".to_vec(), window));
        }
    }
    out
}

/// `FT.AGGREGATE index query [options]`.
///
/// The same two halves as a search, with the same seam between them. What is
/// different is on either side of it: nothing sorts the answer, so the rows come
/// back in the order the index holds them, and a row is a list of properties
/// rather than a key with its fields hung off it.
pub(super) fn roll(server: &Server, db: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
    aggregated(server, db, args, 2, None, out)
}

/// The same, with the query somewhere other than the third word and with
/// somebody watching, which is how `FT.PROFILE` runs an aggregation.
pub(super) fn aggregated(
    server: &Server,
    db: usize,
    args: Args<'_>,
    at: usize,
    watch: Option<&mut Watch>,
    out: &mut Out,
) -> Result<()> {
    let mut watch = watch;
    let name = args.get(1);
    let query = args.get(at);
    let clock = Instant::now();
    let asked;
    let mut canon: Box<[u8]> = Box::default();
    let (total, rows) = {
        let mut reg = server.search.lock();
        let Some(index) = reg.open(name) else {
            Fail::naming(MISSING, name).write(out);
            return Ok(());
        };
        // The first of the two passes, which reads every word and looks no
        // property up. All it is really for is the handful of words that say
        // how the query itself is read, and it is a whole pass rather than a
        // scan for those four because every other word has to be checked
        // before the query is, which is measured.
        let read = match options(args, at + 1, Mode::Aggregate, index, Bind::Reading) {
            Ok(read) => read,
            Err(text) => {
                out.error(&text);
                return Ok(());
            }
        };
        let ask = Ask {
            dialect: read.dialect,
            params: &read.params,
            verbatim: read.verbatim,
            stopwords: read.stopwords,
        };
        let node = match query::parse(query, index, &ask) {
            Ok(node) => node,
            Err(bad) => {
                out.error(&refused(&bad));
                return Ok(());
            }
        };
        // The second pass, with the distances the query yields already on the
        // row, so a step naming one of them finds it where a step naming a
        // field of the key finds that.
        let held = query::yields(&node);
        asked = match options(args, at + 1, Mode::Aggregate, index, Bind::Binding(&held)) {
            Ok(mut asked) => {
                asked.rows.nearest = held
                    .iter()
                    .find(|want| want.ordered)
                    .map(|want| want.name.clone());
                asked.rows.distance = held;
                asked
            }
            Err(text) => {
                out.error(&text);
                return Ok(());
            }
        };
        if asked.cursor.is_some() {
            canon = index.name.clone();
            // Asked before the query runs, because a real server takes the
            // cursor's place in the table before it makes the reply that goes in
            // it, so an index holding its hundred and twenty eight refuses even
            // a cursor that would close on its first chunk.
            if let Err(fail) = cursor::room(server, &canon, 1) {
                fail.write(out);
                return Ok(());
            }
        }
        let order = match buffered(&asked) {
            true => Order::Backwards,
            false => Order::Forwards,
        };
        if let Some(watch) = watch.as_deref_mut() {
            watch.parsing = clock.elapsed();
        }
        let shaped = shape(node, index, &asked.rows);
        if let Some(watch) = watch.as_deref_mut() {
            watch.creating = clock.elapsed() - watch.parsing;
        }
        let (total, rows, ran) = gather(
            index,
            shaped,
            &asked.rows,
            order,
            !asked.pipe.steps.is_empty(),
            watch.is_some(),
        );
        if let Some(watch) = watch.as_deref_mut() {
            if let Some((ran, spent)) = ran {
                watch.ran = Some(ran);
                watch.walking = spent;
            }
            watch.steps.push((b"Index".to_vec(), total));
            if !asked.rows.distance.is_empty() {
                watch.steps.push((b"Metrics Applier".to_vec(), total));
            }
        }
        (total, rows)
    };
    rolled(server, db, total, &rows, &asked, &canon, watch, out);
    Ok(())
}

/// Reads the keys the pipeline asked for and writes the reply.
///
/// The count at the front is not the number of documents that answered, except
/// when it is. What it really reports is how far the reply has got through
/// them, which is why it changes with the protocol and with whether anything
/// asked for a field. The four cases are set out where they are worked out.
#[allow(clippy::too_many_arguments)]
fn rolled(
    server: &Server,
    db: usize,
    total: usize,
    rows: &[Row],
    asked: &Asked<'_>,
    index: &[u8],
    watch: Option<&mut Watch>,
    out: &mut Out,
) {
    if !asked.pipe.steps.is_empty() {
        piped(server, db, total, rows, asked, index, watch, out);
        return;
    }
    let pipe = &asked.pipe;
    let want = &asked.rows;
    let held: Vec<Option<indexing::Document>> = match pipe.loader {
        true => rows
            .iter()
            .map(|row| indexing::read(&server.dbs[db], &row.key))
            .collect(),
        false => Vec::new(),
    };
    let mut built: Vec<Rolled<'_>> = Vec::with_capacity(rows.len());
    let mut lost = 0;
    for (at, row) in rows.iter().enumerate() {
        if pipe.loader {
            // A key that answered the query and is no longer there takes its
            // row out of the reply and one off the count, the same way a search
            // does. The walk still went through it, so it widens the gap in
            // front of whatever row comes next.
            let Some(doc) = held.get(at).and_then(Option::as_ref) else {
                lost += 1;
                continue;
            };
            built.push((row, props(doc, pipe)));
        } else {
            built.push((row, Vec::new()));
        }
    }
    let total = total - lost;
    // The same two steps a pipeline reports, for a pipeline that has no steps
    // of its own. A bare `LOAD` still opens the key it names.
    if let Some(watch) = watch {
        if pipe.addscores {
            watch.steps.push((b"Scorer".to_vec(), built.len()));
        }
        if pipe.loads {
            watch.steps.push((b"Loader".to_vec(), built.len()));
        }
    }
    if let Some(asks) = asked.cursor {
        // A cursor holds the rows rather than the documents they were read
        // from, so what is handed over here is a copy. `LIMIT 0 0` is the one
        // shape where there is nothing to copy and a number all the same.
        let made = Made::Rolled(
            built
                .into_iter()
                .map(|(row, props)| (row.clone(), owned(props)))
                .collect(),
        );
        let kept = Kept {
            made,
            // Nothing here throws a row away, so every document the query
            // answered is a document that reached the window.
            walk: vec![true; total],
            shows: asked.rolls(),
            total,
            whole: want.count == 0 || buffered(asked),
            loader: pipe.loader,
            offset: want.offset,
            window: want.count,
        };
        cursor::open(server, index, kept, asks, out);
        return;
    }
    let wide = out.proto().is_resp3();
    // `LIMIT 0 0` is a client asking for the count and nothing else, and it gets
    // the real one. So does one whose scorer had to see the whole answer before
    // any of it could be written. Everything else reports how far the reply
    // reached, which under RESP2 is one row, because the array header and the
    // number in front of it go on the wire as soon as the first row exists.
    let whole = want.count == 0 || buffered(asked);
    let count = match whole {
        true => total,
        false => {
            let reached = match wide {
                true => built.len(),
                false => 1,
            };
            want.offset.saturating_add(reached).min(total)
        }
    };
    rolls(count, &built, asked.rolls(), out);
}

/// The rows of an aggregation that ran no pipeline step, on either protocol.
fn rolls(count: usize, built: &[Rolled<'_>], shows: Shows, out: &mut Out) {
    if out.proto().is_resp3() {
        rolled_deep(count, built, shows, out);
        return;
    }
    // The count, then a fixed number of elements for every row: whichever of
    // the score, the payload and the sort key were asked for, then the
    // properties unless `NOCONTENT` took them away. A `NOCONTENT` with nothing
    // else on it leaves an array of one.
    let per = usize::from(shows.scores)
        + usize::from(shows.payloads)
        + usize::from(shows.sortkeys)
        + usize::from(shows.fields);
    out.array(1 + built.len() * per);
    out.int(count as i64);
    for (row, fields) in built {
        if shows.scores {
            out.double(row.score);
        }
        if shows.payloads {
            match &row.payload {
                Some(payload) => out.bulk(payload),
                None => out.nil(),
            }
        }
        if shows.sortkeys {
            // Always null, because nothing here sorted: a pipeline with a
            // `SORTBY` in it is written by the step runner instead.
            out.nil();
        }
        if shows.fields {
            out.map(fields.len() + usize::from(shows.addscores) + row.dists.len());
            spaced(row, out);
            if shows.addscores {
                out.bulk(b"__score");
                out.bulk(twelve(row.score).as_bytes());
            }
            for (field, value) in fields {
                out.bulk(field);
                out.bulk(value);
            }
        }
    }
}

/// The distances a vector clause measured, which lead the properties of a row
/// the way they lead the fields of a search reply.
///
/// They are written here rather than read onto the row like everything else
/// because nothing read them off the key: an aggregation with no step in it
/// never opens a key at all and still answers them.
fn spaced(row: &Row, out: &mut Out) {
    for (name, away) in &row.dists {
        out.bulk(name);
        out.bulk(twelve(*away).as_bytes());
    }
}

/// The RESP3 shape, which is the same map of five a search answers with.
///
/// A row is a map too, and the only thing missing from it beside a search is
/// the `id`: an aggregation is about the properties and not about the keys they
/// came off.
fn rolled_deep(count: usize, built: &[Rolled<'_>], shows: Shows, out: &mut Out) {
    out.map(5);
    out.simple(b"attributes");
    out.array(0);
    out.simple(b"format");
    out.simple(b"STRING");
    out.simple(b"results");
    out.array(built.len());
    for (row, fields) in built {
        out.map(
            1 + usize::from(shows.scores)
                + usize::from(shows.payloads)
                + usize::from(shows.sortkeys)
                + usize::from(shows.fields),
        );
        if shows.scores {
            out.simple(b"score");
            out.double(row.score);
        }
        if shows.payloads {
            out.simple(b"payload");
            match &row.payload {
                Some(payload) => out.bulk(payload),
                None => out.nil(),
            }
        }
        if shows.sortkeys {
            out.simple(b"sortkey");
            out.nil();
        }
        if shows.fields {
            out.simple(b"extra_attributes");
            out.map(fields.len() + usize::from(shows.addscores) + row.dists.len());
            spaced(row, out);
            if shows.addscores {
                out.bulk(b"__score");
                out.bulk(twelve(row.score).as_bytes());
            }
            for (field, value) in fields {
                out.bulk(field);
                out.bulk(value);
            }
        }
        out.simple(b"values");
        out.array(0);
    }
    out.simple(b"total_results");
    out.int(count as i64);
    out.simple(b"warning");
    out.array(0);
}

/// The properties of one key, under the names the client asked for them under.
///
/// The named loads come first in the order they were named, then `LOAD *` adds
/// whatever the key holds that has not been named yet. A name is only answered
/// once, so `LOAD 2 @n @n` sends one property, and a rename does not stand in
/// for the field it renamed, so `LOAD 3 @t AS tt LOAD *` sends both `tt` and
/// `t`. A field the key does not hold is left out rather than sent empty.
fn props<'d, 'w: 'd>(doc: &'d indexing::Document, pipe: &Pipe<'w>) -> Vec<(&'d [u8], &'d [u8])> {
    let pairs = doc.pairs();
    let mut out: Vec<(&[u8], &[u8])> = Vec::with_capacity(pipe.load.len());
    for (field, name) in &pipe.load {
        if out.iter().any(|(held, _)| held == name) {
            continue;
        }
        let Some((_, value)) = pairs.iter().find(|(held, _)| held == field) else {
            continue;
        };
        out.push((*name, *value));
    }
    if pipe.all {
        for (field, value) in pairs {
            if !out.iter().any(|(held, _)| *held == field) {
                out.push((field, value));
            }
        }
    }
    out
}

/// A double written the way `%.12g` writes it, which is how a `__score` goes on
/// the wire.
///
/// Twelve significant digits and not seventeen, which is measured: a score of
/// 0.9343092373768334 is answered as `0.934309237377` under `ADDSCORES` and in
/// full beside `WITHSCORES`, so the two are not the same formatter. The
/// spelling lives beside the expression language, because every number an
/// expression answers goes on the wire the same way.
use yo_search::expr::{seventeen, twelve};

/// What a field of the key becomes when it is read onto a pipeline row, which
/// is a number when the schema says the field holds one.
fn holds(index: &Index, field: &[u8]) -> Shape {
    let numeric = index
        .field(field)
        .is_some_and(|held| held.kind == Kind::Numeric);
    match numeric {
        true => Shape::Number,
        false => Shape::Words,
    }
}

/// The query with everything the arguments outside it asked for hung on it.
///
/// The order matters. The slop and the order go on the query the client wrote,
/// because they are the same thing an attribute clause hangs on a group and
/// they are about the words in that group. The filters go on the outside of
/// that, in a wrapper of their own, so a range written as `FILTER` does not
/// join the group whose words are being counted and change what the slop means.
fn shape(node: Node, index: &Index, rows: &Rows<'_>) -> Node {
    let mut node = node;
    if let Some(want) = &rows.infields {
        // A field nobody knows contributes no bit, so `INFIELDS 1 nope` asks
        // for no field at all and answers nothing, and `INFIELDS 0` narrows
        // nothing because there was no list to narrow to.
        if !want.is_empty() {
            let mask = want
                .iter()
                .filter_map(|field| query::explain::bit(index, field))
                .fold(0 as Mask, |mask, bit| mask | bit);
            node.narrow(mask);
        }
    }
    node.slop = rows.slop;
    node.inorder = rows.inorder || node.inorder;
    if rows.filters.is_empty() && rows.circles.is_empty() {
        return node;
    }
    // A filter beside a bare `*` is the whole query, because asking for every
    // document and then asking for the ones in a range is asking for the ones
    // in the range. Measured, and visible in the score: under `TFIDF` the
    // query `*` with one `FILTER` scores one where two filters score two, so
    // the wildcard is not there to be counted.
    let mut under = match node.what {
        What::Wildcard => Vec::new(),
        _ => vec![node],
    };
    for (field, min, max) in &rows.filters {
        under.push(Node::new(What::Numeric(Range {
            field: (*field).into(),
            min: *min,
            max: *max,
            min_open: false,
            max_open: false,
        })));
    }
    for circle in &rows.circles {
        under.push(Node::new(What::Geo(circle.clone())));
    }
    Node::new(What::Intersect(under))
}

/// Walks the query, scores what it found, sorts it and cuts out the window.
///
/// The total comes back beside the window because the window is not the whole
/// of what answered, and because `LIMIT 0 0` is a client asking for the total
/// and nothing else.
fn gather(
    index: &Index,
    node: Node,
    rows: &Rows<'_>,
    order: Order,
    whole: bool,
    profiling: bool,
) -> (usize, Vec<Row>, Option<(walk::Ran, Duration)>) {
    let facts = index.held.facts();
    // How far apart the words landed is worked out only for the three scorers
    // that divide by it, because it costs a pass over the places of every
    // document that answered and the scorer nobody names is not one of them.
    let measure = rows.scorer.divides();
    let mut ran = None;
    let mut spent = Duration::ZERO;
    let walked = if profiling {
        let clock = Instant::now();
        let (walked, tree) = walk::profiled(&index.held, &node, measure);
        spent = clock.elapsed();
        ran = Some(tree);
        walked
    } else if measure {
        walk::spaced(&index.held, &node)
    } else {
        walk::run(&index.held, &node)
    };
    // What each distance the query asked for was measured against, looked up
    // once rather than once a row. A field the schema does not hold vectors for
    // is not one of these, which is a query naming a field that went away.
    let asking: Vec<(&Yield, &yo_search::vecs::Vecs)> = rows
        .distance
        .iter()
        .filter_map(|want| Some((want, index.held.vecs(&want.field)?)))
        .collect();
    let mut found: Vec<Scored<'_>> = walked
        .into_iter()
        .filter_map(|hit| {
            let doc = index.held.docs.get(hit.id)?;
            // `INKEYS` is a filter on the answer and not on the query, and it
            // comes off before the total is taken, which is measured:
            // `INKEYS 1 d:1` over a query that answers three keys answers a
            // total of one.
            if let Some(keys) = &rows.inkeys
                && !keys.contains(&&*doc.key)
            {
                return None;
            }
            let score = rows
                .scorer
                .of(&facts, doc, &hit.found, rows.payload, hit.slop);
            // Worked out here because a `SORTBY` may be about to sort on one of
            // them, so they cannot wait for the window the way the marking up
            // of a row does.
            let dists = asking
                .iter()
                .filter_map(|(want, vecs)| {
                    let away = vecs.distance(hit.id, &want.asked)?;
                    Some((want.name.clone(), f64::from(away)))
                })
                .collect();
            Some((hit, score, dists))
        })
        .collect();
    // The one scorer that cannot finish a document at a time, because what it
    // divides by is the best score in the whole answer.
    let mut scores: Vec<f64> = found.iter().map(|(_, score, _)| *score).collect();
    // Taken before the settling rather than after, because it is what the
    // settling divides by and an explanation prints it.
    let best = scores.iter().copied().fold(0.0_f64, f64::max);
    rows.scorer.settle(&mut scores);
    for (row, score) in found.iter_mut().zip(&scores) {
        row.1 = *score;
    }
    // A `SORTBY` over a field the index keeps a copy of, which is sorted here
    // and windowed here. A `SORTBY` over a field it keeps no copy of is not:
    // the value is in the key, the keyspace is not locked here and the registry
    // is, and a writer takes those two the other way round. So that one hands
    // the whole answer back and is sorted once the lock has been let go of.
    let by = rows
        .sorting
        .as_ref()
        .filter(|by| by.slot.is_some() || by.distance);
    // Whether a sort is still to come, which is what the window and the order
    // below both turn on.
    let later = rows.sorting.is_some() && by.is_none();
    // Best first, and a tie goes to the document that was written first. A
    // `NaN` sorts last rather than poisoning the comparison. An aggregation
    // with nothing sorting it hands the documents back in the order the index
    // holds them, which is measured: a five document answer whose scores run
    // 0.93, 0.69, 0.85, 0.27, 0.24 comes back in exactly that order, and those
    // five documents are in ascending document number.
    match (by, order) {
        // A sort by a field, where a tie goes to the document written first and
        // is turned over with everything else by `DESC`: a descending answer is
        // the ascending one backwards, apart from the rows with no value at
        // all, which are last either way.
        // A sort by a distance the query yielded, which is a number this
        // gathering worked out rather than anything the key holds, so it never
        // reaches the slower sort that reads the keys back.
        (Some(by), _) if by.distance => {
            found.sort_by(|a, b| {
                let ids = match by.desc {
                    true => b.0.id.cmp(&a.0.id),
                    false => a.0.id.cmp(&b.0.id),
                };
                let held = |row: &Scored<'_>| {
                    row.2
                        .iter()
                        .find(|(name, _)| **name == *by.field.attribute)
                        .map(|(_, away)| Sorted::Number(*away))
                };
                sorted::order(held(a).as_ref(), held(b).as_ref(), by.desc).then(ids)
            });
        }
        (Some(by), _) => {
            let slot = by.slot.unwrap_or_default();
            let held = |id: u32| index.held.docs.get(id).and_then(|doc| doc.sorted(slot));
            found.sort_by(|a, b| {
                let ids = match by.desc {
                    true => b.0.id.cmp(&a.0.id),
                    false => a.0.id.cmp(&b.0.id),
                };
                sorted::order(held(a.0.id), held(b.0.id), by.desc).then(ids)
            });
        }
        // A sort that has not run yet, which is sorted once the lock has been
        // let go of. It goes back in document number order so that its ties come
        // out the same way round as the ties of a sort that ran here, since the
        // sort that runs later is stable and keeps whatever order it was handed.
        (None, _) if later => found.sort_by_key(|(hit, _, _)| hit.id),
        (None, Order::Ranked) => found.sort_by(|a, b| {
            b.1.partial_cmp(&a.1)
                .unwrap_or(core::cmp::Ordering::Equal)
                .then(a.0.id.cmp(&b.0.id))
        }),
        // An aggregation over a nearest neighbour clause, whose rows arrive
        // nearest first and stay that way because no step has sorted them. A
        // tie goes to the document written first, which is measured: a query
        // sitting between two documents answers the lower number of the two
        // ahead of the higher.
        //
        // Written as a match inside the arm rather than as an `if let` guard on
        // it. The guard reads better and says the same thing, but `if_let_guard`
        // is not stable in 1.94, which is what `rust-version` says and what the
        // msrv job builds with.
        (None, Order::Forwards) => match &rows.nearest {
            Some(name) => {
                let away = |row: &Scored<'_>| {
                    row.2
                        .iter()
                        .find(|(held, _)| **held == **name)
                        .map_or(f64::INFINITY, |(_, away)| *away)
                };
                found.sort_by(|a, b| {
                    away(a)
                        .partial_cmp(&away(b))
                        .unwrap_or(core::cmp::Ordering::Equal)
                        .then(a.0.id.cmp(&b.0.id))
                });
            }
            None => found.sort_by_key(|(hit, _, _)| hit.id),
        },
        (None, Order::Backwards) => {
            found.sort_by_key(|(hit, _, _)| core::cmp::Reverse(hit.id));
        }
    }
    let total = found.len();
    // A grouping step folds every document that answered and the window goes on
    // what came out of it, so the window is not applied here at all. Neither is
    // it applied to a sort that has not happened yet.
    let (offset, count) = match whole || later {
        true => (0, usize::MAX),
        false => (rows.offset, rows.count),
    };
    let why = Why::new(rows.scorer, facts).about(rows.payload).over(best);
    let window = found
        .into_iter()
        .skip(offset)
        .take(count)
        .filter_map(|(hit, score, dists)| {
            let doc = index.held.docs.get(hit.id)?;
            let sort = match by {
                // The value a distance sort compared, which goes on the row the
                // same way the value a field sort compared goes on it.
                Some(by) if by.distance => dists
                    .iter()
                    .find(|(name, _)| **name == *by.field.attribute)
                    .map(|(_, away)| Sorted::Number(*away)),
                Some(by) => doc.sorted(by.slot.unwrap_or_default()).cloned(),
                None => None,
            };
            Some(Row {
                key: doc.key.clone(),
                score,
                payload: doc.payload.clone(),
                sort,
                note: rows.explaining.then(|| why.note(doc, &hit.found, hit.slop)),
                dists,
            })
        })
        .collect();
    (total, window, ran.map(|ran| (ran, spent)))
}

/// A row that survived the read, with the fields the reply wants off its key.
///
/// The fields borrow the document they were read from, which is why these are
/// kept apart from the rows that go on the wire: a `SUMMARIZE` rewrites some of
/// the values and the rewritten copies have to outlive the rows pointing at
/// them.
type Held<'a> = (&'a Row, Option<Vec<(&'a [u8], &'a [u8])>>);

/// Reads the keys in the window and writes the reply.
///
/// The reading happens first and all of it, because a key that will not read
/// takes its row out of the reply and one off the total, and the total is the
/// first thing on the wire under RESP2.
fn write(
    server: &Server,
    db: usize,
    total: usize,
    rows: &[Row],
    asked: &Asked<'_>,
    index: &[u8],
    out: &mut Out,
) {
    let want = &asked.rows;
    let loading = want.loading();
    // A sort by a field the index keeps no copy of is finished here, because
    // the value is in the key and the key could not be read under the lock. So
    // the whole answer arrived rather than a window on it, and every row of it
    // has its key read whether or not the client asked for any fields.
    let slow = want
        .sorting
        .as_ref()
        .is_some_and(|by| by.slot.is_none() && !by.distance);
    let mut held: Vec<Option<indexing::Document>> = if loading || slow {
        rows.iter()
            .map(|row| indexing::read(&server.dbs[db], &row.key))
            .collect()
    } else {
        Vec::new()
    };
    let mut lost = 0;
    let carried: Vec<Row>;
    let mut rows = rows;
    if let Some(by) = want
        .sorting
        .as_ref()
        .filter(|by| by.slot.is_none() && !by.distance)
    {
        let mut pairs: Vec<(Row, Option<indexing::Document>)> = rows
            .iter()
            .cloned()
            .zip(core::mem::take(&mut held))
            // A key that answered the query and went away before it could be
            // read is out of the answer and off the total, the same as it is
            // when nothing sorted.
            .filter(|(_, doc)| doc.is_some())
            .collect();
        lost = rows.len() - pairs.len();
        for (row, doc) in &mut pairs {
            row.sort = doc
                .as_ref()
                .and_then(|doc| doc.held(&by.field.identifier))
                .and_then(|raw| Sorted::read(&by.field, raw));
        }
        // The rows arrived in document number order, so turning them over first
        // and sorting them with a sort that keeps what it does not have to move
        // gives a descending answer the ascending one's tie order backwards,
        // which is what the index's own sort does.
        if by.desc {
            pairs.reverse();
        }
        pairs.sort_by(|a, b| sorted::order(a.0.sort.as_ref(), b.0.sort.as_ref(), by.desc));
        (carried, held) = pairs.into_iter().skip(want.offset).take(want.count).unzip();
        rows = &carried;
    }
    // The value the sort compared, written out once per row, because the fields
    // of a row point at what they were read from and this one is worked out.
    let keys: Vec<Option<Vec<u8>>> = match want.sorting.is_some() {
        true => rows.iter().map(|row| shown(row.sort.as_ref())).collect(),
        false => Vec::new(),
    };
    // The distances, written out once per row for the same reason: they are
    // worked out rather than read, so the bytes have to live somewhere the
    // fields of a row can point at.
    let aways: Vec<Pairs> = rows
        .iter()
        .map(|row| {
            row.dists
                .iter()
                .map(|(name, away)| (name.clone(), twelve(*away).into_bytes().into()))
                .collect()
        })
        .collect();
    // The rows that survived the read, with their fields as the key holds them.
    // Kept apart from the reply rows below because a `SUMMARIZE` rewrites some
    // of these values and the rewritten copies have to outlive the rows that
    // point at them.
    let mut kept: Vec<Held<'_>> = Vec::with_capacity(rows.len());
    for (at, row) in rows.iter().enumerate() {
        if !loading {
            kept.push((row, None));
            continue;
        }
        let Some(doc) = held.get(at).and_then(Option::as_ref) else {
            lost += 1;
            continue;
        };
        let key = keys.get(at).and_then(Option::as_deref);
        let away = aways.get(at).map_or(&[][..], Pairs::as_slice);
        kept.push((row, Some(pick(doc, want, key, away))));
    }
    let redone = marked(&kept, want);
    let mut built: Vec<Built<'_>> = kept
        .iter()
        .zip(&redone)
        .map(|((row, fields), redone)| {
            let fields = fields.as_ref().map(|fields| {
                fields
                    .iter()
                    .zip(redone)
                    .map(|((name, value), redone)| match redone {
                        Some(redone) => (*name, &**redone),
                        None => (*name, *value),
                    })
                    .collect()
            });
            (*row, fields)
        })
        .collect();
    shared(&mut built, &want.upfront);
    let total = total - lost;
    if let Some(asks) = asked.cursor {
        // A search is settled: it ranked the whole answer before it wrote a row
        // of it, so the first chunk carries the real total and the rest carry
        // nought, and none of the walk arithmetic comes into it.
        let made = Made::Found(
            built
                .into_iter()
                .map(|(row, fields)| (row.clone(), fields.map(owned)))
                .collect(),
        );
        let kept = Kept {
            made,
            walk: Vec::new(),
            shows: want.found(),
            total,
            whole: true,
            loader: loading,
            offset: want.offset,
            window: want.count,
        };
        cursor::open(server, index, kept, asks, out);
        return;
    }
    found(total, &built, want.found(), out);
}

/// What a `SUMMARIZE` and a `HIGHLIGHT` made of every value on its way out.
///
/// One slot per field of every row, holding nothing where the value goes out as
/// the key holds it. The slots are kept rather than the rows being rewritten in
/// place because a row points at the document it was read from, and a value
/// that has been cut down or marked is neither.
///
/// A field the schema does not index is cut down but never marked, and never
/// treated as holding a match either, which is why an unindexed field of a hash
/// comes back as its own front however much of the query is written across it.
fn marked(kept: &[Held<'_>], want: &Rows<'_>) -> Vec<Vec<Option<Vec<u8>>>> {
    let widths = || {
        kept.iter()
            .map(|(_, fields)| vec![None; fields.as_ref().map_or(0, Vec::len)])
            .collect()
    };
    if want.summarize.is_none() && want.highlight.is_none() {
        return widths();
    }
    let mut english = English::new();
    let nothing = Wanted::default();
    let stops = want.stops.as_deref();
    let wanted = &want.wanted;
    kept.iter()
        .map(|(_, fields)| {
            fields
                .iter()
                .flatten()
                .map(|(name, value)| {
                    // A distance is worked out rather than read off the key, so
                    // neither clause touches it: a `SUMMARIZE` over every field
                    // leaves `__v_score` whole.
                    if want.distance.iter().any(|held| *held.name == **name) {
                        return None;
                    }
                    let named = |list: &[&[u8]]| list.contains(name);
                    // Once any of the two clauses has named a field, the fields
                    // neither of them named are left alone entirely, even by the
                    // clause that named nothing and so covers everything. That
                    // is why `SUMMARIZE HIGHLIGHT FIELDS 1 a` cuts nothing down
                    // but `SUMMARIZE HIGHLIGHT` cuts every field down.
                    if !want.upfront.is_empty() && !named(&want.upfront) {
                        return None;
                    }
                    let text = want.texts.iter().any(|held| **held == **name);
                    let touches = |list: &Vec<&[u8]>| list.is_empty() || named(list);
                    let wrap = want
                        .highlight
                        .as_ref()
                        .filter(|(list, _)| touches(list))
                        .map(|(_, wrap)| wrap);
                    let seen = match text {
                        true => wanted,
                        false => &nothing,
                    };
                    match want.summarize.as_ref().filter(|(list, _)| touches(list)) {
                        Some((_, trim)) => Some(summary::summarize(
                            value,
                            seen,
                            stops,
                            trim,
                            wrap,
                            &mut english,
                        )),
                        None => wrap
                            .map(|wrap| summary::highlight(value, seen, stops, wrap, &mut english)),
                    }
                })
                .collect()
        })
        .collect()
}

/// Puts every row of a window in the one order the whole window shares.
///
/// A real server does not answer a key in that key's own order. It keeps one
/// ordered list of names for the answer, puts a name on the end of it the first
/// time a row carries one, and then writes each row in that list's order,
/// leaving out the names that row has nothing under. So the first row of a
/// window fixes the order of the fields it holds and every row after it can only
/// add to the end, which is why the same key comes back one way at `LIMIT 0 7`
/// and another way at `LIMIT 1 2`, and why a sort puts its own field in front of
/// everything: the sort registers its name before any row is read.
///
/// `upfront` is the names a `SUMMARIZE FIELDS` or a `HIGHLIGHT FIELDS` list
/// registered, which happens while the clause is being read and so before any
/// row is read as well. That is the whole of why naming a field in one of those
/// lists moves it to the front of the reply.
fn shared<'a>(built: &mut [Built<'a>], upfront: &[&'a [u8]]) {
    let mut order: Vec<&[u8]> = upfront.to_vec();
    for (_, fields) in built.iter() {
        for (name, _) in fields.iter().flatten() {
            if !order.contains(name) {
                order.push(name);
            }
        }
    }
    for (_, fields) in built.iter_mut() {
        let Some(fields) = fields else {
            continue;
        };
        let mut out = Vec::with_capacity(fields.len());
        for name in &order {
            if let Some((_, value)) = fields.iter().find(|(held, _)| held == name) {
                out.push((*name, *value));
            }
        }
        *fields = out;
    }
}

/// Copies a row's fields out of the documents they were read from.
///
/// Only a cursor needs this. A reply that goes out in one piece points at the
/// documents, which are alive for as long as writing it takes.
fn owned(pairs: Vec<(&[u8], &[u8])>) -> Pairs {
    pairs
        .into_iter()
        .map(|(name, value)| (name.into(), value.into()))
        .collect()
}

/// A row's score, with the explanation beside it when one was asked for.
///
/// Measured on both protocols: the score element becomes an array of two, the
/// score exactly as it would have gone out on its own and then the tree. It is
/// not a third element beside the score, so a client that asked for the
/// explanation reads the score out of a pair.
fn worth(row: &Row, out: &mut Out) {
    let Some(note) = &row.note else {
        out.double(row.score);
        return;
    };
    out.array(2);
    out.double(row.score);
    reason(note, out);
}

/// One line of an explanation and everything under it.
///
/// A line with nothing under it is a string, and a line with children is an
/// array of two holding the line and the list. The list is one level of array
/// on its own rather than the children being spread into the pair, which is
/// what makes the whole thing readable by walking pairs.
fn reason(note: &Note, out: &mut Out) {
    match note {
        Note::Line(line) => out.bulk(line.as_bytes()),
        Note::Under(line, under) => {
            out.array(2);
            out.bulk(line.as_bytes());
            out.array(under.len());
            for note in under {
                reason(note, out);
            }
        }
    }
}

/// The rows of a search on the wire, on either protocol.
fn found(total: usize, built: &[Built<'_>], shows: Shows, out: &mut Out) {
    if out.proto().is_resp3() {
        deep(total, built, shows, out);
        return;
    }
    // One element for the total, then a fixed number for every row: the key,
    // then whichever of the score, the payload and the fields were asked for.
    let per = 1
        + usize::from(shows.scores)
        + usize::from(shows.payloads)
        + usize::from(shows.sortkeys)
        + usize::from(shows.fields);
    out.array(1 + built.len() * per);
    out.int(total as i64);
    for (row, fields) in built {
        out.bulk(&row.key);
        if shows.scores {
            worth(row, out);
        }
        if shows.payloads {
            match &row.payload {
                Some(payload) => out.bulk(payload),
                None => out.nil(),
            }
        }
        if shows.sortkeys {
            match sortkey(row.sort.as_ref()) {
                Some(sorted) => out.bulk(&sorted),
                None => out.nil(),
            }
        }
        if let Some(fields) = fields {
            out.map(fields.len());
            for (field, value) in fields {
                out.bulk(field);
                out.bulk(value);
            }
        }
    }
}

/// The RESP3 shape, which is a map of five and not an array of anything.
///
/// The five keys are always all five and always in this order, even for an
/// answer with nothing in it, and `attributes`, `warning` and every row's
/// `values` are always empty. That is measured rather than assumed: they are
/// there for `FT.AGGREGATE` and a client that reads them finds them.
fn deep(total: usize, built: &[Built<'_>], shows: Shows, out: &mut Out) {
    out.map(5);
    out.simple(b"attributes");
    out.array(0);
    out.simple(b"format");
    out.simple(b"STRING");
    out.simple(b"results");
    out.array(built.len());
    for (row, fields) in built {
        out.map(
            2 + usize::from(shows.scores)
                + usize::from(shows.payloads)
                + usize::from(shows.sortkeys)
                + usize::from(fields.is_some()),
        );
        out.simple(b"id");
        out.bulk(&row.key);
        if shows.scores {
            out.simple(b"score");
            worth(row, out);
        }
        if shows.payloads {
            out.simple(b"payload");
            match &row.payload {
                Some(payload) => out.bulk(payload),
                None => out.nil(),
            }
        }
        if shows.sortkeys {
            out.simple(b"sortkey");
            match sortkey(row.sort.as_ref()) {
                Some(sorted) => out.bulk(&sorted),
                None => out.nil(),
            }
        }
        if let Some(fields) = fields {
            out.simple(b"extra_attributes");
            out.map(fields.len());
            for (field, value) in fields {
                out.bulk(field);
                out.bulk(value);
            }
        }
        out.simple(b"values");
        out.array(0);
    }
    out.simple(b"total_results");
    out.int(total as i64);
    out.simple(b"warning");
    out.array(0);
}

/// The fields of one key that go in the reply, under the names they go under.
///
/// Everything the key holds when the client named no fields, which is every
/// field of the hash and not only the ones in the schema. When it did name
/// them, they come back in the order it named them, a field the key does not
/// hold is left out rather than sent empty, and a name that is not a field at
/// all leaves an empty list rather than an error.
///
/// A sort puts its own field in front of the rest either way. With no `RETURN`
/// the value the sort compared goes in first under the name the client sorted
/// by, and then the key's own fields land on top of it: a field whose name the
/// key holds keeps the place the sort gave it and takes the value the key has,
/// and a field the schema renamed appears twice, once folded under the name the
/// query calls it and once as it was written under the name the key calls it.
/// With a `RETURN` there is nothing to put in, since the sort field is either
/// on the list already or was not asked for, so it is moved to the front
/// instead. All of that is measured.
fn pick<'d, 'w: 'd>(
    doc: &'d indexing::Document,
    want: &'d Rows<'w>,
    key: Option<&'d [u8]>,
    away: &'d [Named],
) -> Vec<(&'d [u8], &'d [u8])> {
    let pairs = doc.pairs();
    let by = want.sorting.as_ref();
    let Some(ret) = &want.ret else {
        let sorted = by.zip(key);
        if sorted.is_none() && away.is_empty() {
            return pairs;
        }
        let mut out: Vec<(&[u8], &[u8])> = Vec::with_capacity(pairs.len() + away.len() + 1);
        for (name, value) in away {
            out.push((&**name, &**value));
        }
        // A sort by a distance named its value once already, so it is not put
        // in twice. A sort by anything else goes in after the distances, which
        // is measured: a `SORTBY n` over a nearest neighbour query answers the
        // distance first and `n` after it.
        if let Some((by, key)) = sorted
            && !out.iter().any(|(held, _)| *held == &*by.field.attribute)
        {
            out.push((&by.field.attribute, key));
        }
        for (name, value) in pairs {
            match out.iter_mut().find(|(held, _)| *held == name) {
                Some(held) => held.1 = value,
                None => out.push((name, value)),
            }
        }
        return out;
    };
    let mut out: Vec<(&[u8], &[u8])> = ret
        .iter()
        .filter_map(|(field, name)| {
            // A distance is on the row before the key is read, so what decides
            // whether it comes back is the name the row would answer under and
            // not the field the value would have been read from. That is why a
            // `RETURN 1 __v_score` answers the distance and a
            // `RETURN 3 __v_score AS x` answers nothing at all: the rename
            // sends the reader to the key, which holds no such field.
            if let Some((_, value)) = away.iter().find(|(held, _)| **held == **name) {
                return Some((&**name, &**value));
            }
            let (_, value) = pairs.iter().find(|(held, _)| *held == &**field)?;
            Some((&**name, *value))
        })
        .collect();
    if let Some(by) = by
        && let Some(at) = out
            .iter()
            .position(|(name, _)| *name == &*by.field.attribute)
    {
        let held = out.remove(at);
        out.insert(0, held);
    }
    // The distances go in front of all of it, in the order the query yielded
    // them and whatever order the `RETURN` named them in.
    let mut front: Vec<(&[u8], &[u8])> = Vec::with_capacity(away.len());
    for (name, _) in away {
        if let Some(at) = out.iter().position(|(held, _)| *held == &**name) {
            front.push(out.remove(at));
        }
    }
    front.append(&mut out);
    front
}

/// The value the sort compared, as it goes on the row beside the other fields.
///
/// A number is written the same twelve significant digits everything else on a
/// row is written to, which is not how the same number goes into a sort key.
fn shown(sort: Option<&Sorted>) -> Option<Vec<u8>> {
    match sort? {
        Sorted::Number(number) => Some(twelve(*number).into_bytes()),
        Sorted::Text(text) => Some(text.to_vec()),
    }
}

/// The sort key element beside a row, when `WITHSORTKEYS` asked for one.
///
/// A number after a hash and text after a dollar, the same as an aggregation
/// writes them, and a null on a row the sort found no value on. The number is
/// written wider than the same number on the row is: a sort key is the value
/// the sort compared and a row holds the value the client reads.
fn sortkey(sort: Option<&Sorted>) -> Option<Vec<u8>> {
    let mut out = Vec::new();
    match sort? {
        Sorted::Number(number) => {
            out.push(b'#');
            out.extend_from_slice(seventeen(*number).as_bytes());
        }
        Sorted::Text(text) => {
            out.push(b'$');
            out.extend_from_slice(text);
        }
    }
    Some(out)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_number_that_is_not_one_is_spelled_the_way_the_wire_spells_it() {
        assert_eq!(twelve(f64::NAN), "nan");
        assert_eq!(twelve(f64::INFINITY), "inf");
        assert_eq!(twelve(f64::NEG_INFINITY), "-inf");
        assert_eq!(twelve(0.0), "0");
        // Twelve significant digits and no more, which is what a `__score` and
        // every reduced number go on the wire as.
        assert_eq!(twelve(0.934_309_237_376_833_4), "0.934309237377");
    }

    #[test]
    fn a_geofilter_reads_a_number_the_way_that_one_option_reads_one() {
        assert_eq!(coord(b"0"), Some(0.0));
        assert_eq!(coord(b"-0.1278"), Some(-0.1278));
        assert_eq!(coord(b"1e2"), Some(100.0));
        // Space in front is allowed and space behind is not.
        assert_eq!(coord(b" 5"), Some(5.0));
        assert_eq!(coord(b"5 "), None);
        // A NaN is a number here and an infinity is not, which is the other way
        // round from every other number the command takes.
        assert!(coord(b"nan").is_some_and(f64::is_nan));
        assert_eq!(coord(b"inf"), None);
        assert_eq!(coord(b"-inf"), None);
        assert_eq!(coord(b"1e400"), None);
        assert_eq!(coord(b"x"), None);
        assert_eq!(coord(b""), None);
    }

    #[test]
    fn a_sort_key_is_written_five_digits_wider_than_the_row_it_sits_beside() {
        use yo_search::expr::seventeen;

        assert_eq!(seventeen(1.0 / 3.0), "0.33333333333333331");
        assert_eq!(seventeen(-4.0), "-4");
        assert_eq!(seventeen(5.5), "5.5");
        // The point where a number turns into a scientific one moves with the
        // width, so this pair is fixed at twelve and split at seventeen.
        assert_eq!(seventeen(1e16), "10000000000000000");
        assert_eq!(seventeen(1e17), "1e+17");
        assert_eq!(seventeen(0.000_01), "1.0000000000000001e-05");
        assert_eq!(seventeen(0.000_1), "0.0001");
    }

    #[test]
    fn a_sort_key_says_which_of_the_two_kinds_of_value_it_is() {
        let number = Sorted::Number(1.0 / 3.0);
        let text = Sorted::Text(b"four score".as_slice().into());
        assert_eq!(
            sortkey(Some(&number)).as_deref(),
            Some(&b"#0.33333333333333331"[..])
        );
        assert_eq!(sortkey(Some(&text)).as_deref(), Some(&b"$four score"[..]));
        assert_eq!(sortkey(None), None);
        // The same value beside the fields of the row is the width every other
        // number on a row is written to, which is five digits narrower.
        assert_eq!(
            shown(Some(&number)).as_deref(),
            Some(&b"0.333333333333"[..])
        );
        assert_eq!(shown(Some(&text)).as_deref(), Some(&b"four score"[..]));
        assert_eq!(shown(None), None);
    }

    #[test]
    fn every_row_of_a_window_is_written_in_the_order_the_window_shares() {
        let row = Row {
            key: Box::default(),
            score: 0.0,
            payload: None,
            note: None,
            sort: None,
            dists: Vec::new(),
        };
        let mut built: Vec<Built<'_>> = vec![
            (
                &row,
                Some(vec![(&b"n"[..], &b"0"[..]), (&b"p"[..], &b"a"[..])]),
            ),
            (
                &row,
                Some(vec![
                    (&b"t"[..], &b"x"[..]),
                    (&b"p"[..], &b"b"[..]),
                    (&b"n"[..], &b"2"[..]),
                ]),
            ),
            (&row, None),
        ];
        shared(&mut built, &[]);
        // The first row fixes the order of what it holds and the second one can
        // only add to the end of it, so `t` lands after `n` and `p` even though
        // the key it was read from holds it first.
        assert_eq!(
            built[1].1.as_deref(),
            Some(
                &[
                    (&b"n"[..], &b"2"[..]),
                    (&b"p"[..], &b"b"[..]),
                    (&b"t"[..], &b"x"[..])
                ][..]
            )
        );
        assert_eq!(built[0].1.as_ref().map(Vec::len), Some(2));
        assert_eq!(built[2].1, None);
    }
}