use yo_common::{Code, Error, Result, parse_i64};
use yo_kv::{Db, Foreign, Keyspace};
use yo_shape::Metric;
use yo_vector::hnsw::Requested;
use yo_vector::{Collection, Match, Quant, Signature, quant};
use super::args::{self, Args};
use super::table::Spec;
use super::vfilter;
use crate::reply::Out;
const NOT_A_VECTOR_SET: &str = "Operation against a key holding the wrong kind of value";
const BAD_VECTOR: &str = "invalid vector specification";
const BAD_COUNT: &str = "COUNT must be a positive integer";
const BAD_EF: &str = "EF must be a positive integer";
const BAD_M: &str = "M must be a positive integer";
const WRONG_QUANT: &str = "asked quantization mismatch with existing vector set";
const BAD_START: &str = "invalid start range format";
const BAD_END: &str = "invalid end range format";
const BACKWARDS: &str = "'-' can only be used as first argument, '+' only as second";
const BAD_COUNT_VALUE: &str = "invalid COUNT value";
const COUNT: usize = 10;
#[derive(Debug)]
pub(super) struct VectorBody {
c: Collection,
asked: Requested,
quant: Quant,
side: Vec<Side>,
}
#[derive(Debug, Default, Clone)]
struct Side {
norm: f32,
range: f32,
attr: Option<Box<[u8]>>,
}
impl Foreign for VectorBody {
fn type_name(&self) -> &'static str {
"vectorset"
}
fn encoding(&self) -> &'static str {
"rabitq"
}
fn memory_bytes(&self) -> usize {
let attrs: usize = self
.side
.iter()
.map(|s| s.attr.as_ref().map_or(0, |a| a.len()))
.sum();
self.c.memory_bytes() + self.side.capacity() * size_of::<Side>() + attrs
}
fn is_empty(&self) -> bool {
self.c.is_empty()
}
}
impl VectorBody {
fn new(dim: usize, quant: Quant, asked: Requested) -> Result<VectorBody> {
let mut c = Collection::new(dim, Metric::L2)?;
c.retune(asked.tuning());
Ok(VectorBody {
c,
asked,
quant,
side: Vec::new(),
})
}
fn embedding(&self, key: &[u8]) -> Option<Vec<f32>> {
let dir = self.c.get(key)?;
Some(quant::restore(self.quant, dir, self.norm(key)))
}
fn norm(&self, key: &[u8]) -> f32 {
match self.c.id(key).and_then(|id| self.side.get(id as usize)) {
Some(s) if s.norm > 0.0 => s.norm,
_ => 1.0,
}
}
fn range(&self, key: &[u8]) -> f32 {
self.c
.id(key)
.and_then(|id| self.side.get(id as usize))
.map_or(0.0, |s| s.range)
}
fn attr(&self, key: &[u8]) -> Option<&[u8]> {
let id = self.c.id(key)?;
self.side.get(id as usize)?.attr.as_deref()
}
fn side_mut(&mut self, key: &[u8]) -> &mut Side {
let id = self.c.id(key).expect("the element was just written") as usize;
if self.side.len() <= id {
self.side.resize(id + 1, Side::default());
}
&mut self.side[id]
}
fn retag(&mut self, key: &[u8]) {
let tag = self.attr(key).map_or(0, vfilter::tag);
self.c.retag(key, tag);
}
fn attributes(&self) -> usize {
self.side.iter().filter(|s| s.attr.is_some()).count()
}
}
fn similarity(q: &[f32], s: &[f32]) -> f64 {
let mut dot = 0.0f32;
for (a, b) in q.iter().zip(s) {
dot = a.mul_add(*b, dot);
}
let cos = dot / (quant::norm(q) * quant::norm(s));
f64::from(((1.0 + cos) / 2.0).clamp(0.0, 1.0))
}
fn score(body: &VectorBody, q: &[f32], hit: &Match) -> f64 {
body.c.get(&hit.key).map_or(0.0, |s| similarity(q, s))
}
pub(super) fn execute(db: &Db, spec: &Spec, args: Args<'_>, out: &mut Out) -> Result<()> {
let mut held = db.hold(args.get(1));
let db = &mut *held;
match spec.name {
"VADD" => vadd(db, args, out),
"VSIM" => vsim(db, args, out),
"VREM" => vrem(db, args, out),
"VCARD" => vcard(db, args, out),
"VDIM" => vdim(db, args, out),
"VEMB" => vemb(db, args, out),
"VINFO" => vinfo(db, args, out),
"VISMEMBER" => vismember(db, args, out),
"VRANDMEMBER" => vrandmember(db, args, out),
"VLINKS" => vlinks(db, args, out),
"VSETATTR" => vsetattr(db, args, out),
"VGETATTR" => vgetattr(db, args, out),
"VRANGE" => vrange(db, args, out),
other => unreachable!("{other} is not a vector set command"),
}
}
fn vadd(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
if args.opt(2).is_some_and(|a| args::is(a, b"reduce")) {
return Err(Error::new(
Code::Unsupported,
"REDUCE is not supported. The vector is stored at the dimension it arrives at, and a projection that quietly changed what VDIM says would be worse than saying so",
));
}
let (v, next) = vector(args, 2)?;
let element = args.opt(next).ok_or_else(args::syntax)?;
let opts = Add::parse(args, next + 1)?;
let body = open(db, args.get(1), v.len(), opts.quant, opts.asked)?;
let squeezed = quant::squeeze(body.quant, &v);
let new = body.c.put(element, &squeezed.dir)?;
let side = body.side_mut(element);
side.norm = squeezed.norm;
side.range = squeezed.range;
if let Some(attr) = opts.attr {
side.attr = Some(attr.into());
}
body.retag(element);
out.bool(new);
Ok(())
}
fn vsim(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let (query, next) = if args::is(args.get(2), b"ele") {
(Query::Element(args.opt(3).ok_or_else(args::syntax)?), 4)
} else {
let (v, next) = vector(args, 2)?;
(Query::Vector(v), next)
};
let opts = Sim::parse(args, next)?;
let Some(body) = read(db, args.get(1))? else {
out.array(0);
return Ok(());
};
let (q, mut hits) = match query {
Query::Element(e) => {
let Some(stored) = body.c.get(e) else {
out.array(0);
return Ok(());
};
let q = stored.to_vec();
let hits = search(body, &q, opts.effort, Some(e), &opts)?;
(q, hits)
}
Query::Vector(v) => {
let q = quant::squeeze(body.quant, &v).dir;
let hits = search(body, &q, opts.effort, None, &opts)?;
(q, hits)
}
};
hits.truncate(opts.count);
answer(body, &q, &hits, &opts, out);
Ok(())
}
fn vrem(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let Some(body) = write(db, args.get(1))? else {
out.bool(false);
return Ok(());
};
let element = args.get(2);
let id = body.c.id(element);
let gone = body.c.remove(element);
if let Some(id) = id.filter(|_| gone)
&& let Some(side) = body.side.get_mut(id as usize)
{
*side = Side::default();
}
out.bool(gone);
db.reap_foreign(args.get(1));
Ok(())
}
fn vcard(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let n = read(db, args.get(1))?.map_or(0, |b| b.c.len());
out.uint(n as u64);
Ok(())
}
fn vdim(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
match read(db, args.get(1))? {
Some(body) => out.uint(body.c.dim() as u64),
None => return Err(Error::new(Code::NotFound, "key does not exist")),
}
Ok(())
}
fn vemb(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let raw = match args.len() {
3 => false,
4 if args::is(args.get(3), b"raw") => true,
_ => return Err(args::syntax()),
};
let Some(body) = read(db, args.get(1))? else {
out.nil_array();
return Ok(());
};
let element = args.get(2);
let Some(v) = body.embedding(element) else {
out.nil_array();
return Ok(());
};
if raw {
let dir = body.c.get(element).expect("the element is there");
let range = body.range(element);
let bytes = quant::raw(body.quant, dir, range);
out.array(if body.quant == Quant::Int8 { 4 } else { 3 });
out.simple(body.quant.token().as_bytes());
out.bulk(&bytes);
out.double(f64::from(body.norm(element)));
if body.quant == Quant::Int8 {
out.double(f64::from(range));
}
return Ok(());
}
out.array(v.len());
for x in v {
out.double(f64::from(x));
}
Ok(())
}
fn vinfo(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let Some(body) = read(db, args.get(1))? else {
out.nil();
return Ok(());
};
let t = body.c.tuning();
let fields: [(&[u8], u64); 8] = [
(b"vector-dim", body.c.dim() as u64),
(b"size", body.c.len() as u64),
(b"attributes-count", body.attributes() as u64),
(b"hnsw-m", body.asked.m as u64),
(b"ef-construction", body.asked.ef_construction as u64),
(b"ef-runtime", body.asked.ef_runtime as u64),
(b"partitions", body.c.partitions() as u64),
(b"code-bytes", body.c.code_bytes() as u64),
];
out.map(fields.len() + 4);
out.bulk(b"index-type");
out.bulk(b"partition");
out.bulk(b"quant-type");
out.bulk(body.quant.token().as_bytes());
for (name, value) in fields {
out.bulk(name);
out.uint(value);
}
out.bulk(b"probe");
out.uint(t.probe as u64);
out.bulk(b"rerank");
out.uint(t.rerank as u64);
Ok(())
}
fn vismember(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let there = read(db, args.get(1))?.is_some_and(|b| b.c.contains(args.get(2)));
out.bool(there);
Ok(())
}
fn vrandmember(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let count = match args.len() {
2 => None,
3 => Some(args.int(2)?),
_ => return Err(args::syntax()),
};
let len = read(db, args.get(1))?.map_or(0, |b| b.c.len());
let Some(count) = count else {
if len == 0 {
out.nil();
return Ok(());
}
let pick = (db.random() % len as u64) as usize;
let body = read(db, args.get(1))?.expect("the set is still there");
out.bulk(body.c.key_at(pick).expect("the draw was under the length"));
return Ok(());
};
if len == 0 {
out.array(0);
return Ok(());
}
let picks = draws(db, count, len);
let body = read(db, args.get(1))?.expect("the set is still there");
out.array(picks.len());
for at in picks {
out.bulk(body.c.key_at(at).expect("the draw was under the length"));
}
Ok(())
}
fn vlinks(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let withscores = match args.len() {
3 => false,
4 if args::is(args.get(3), b"withscores") => true,
_ => return Err(args::syntax()),
};
let Some(body) = read(db, args.get(1))? else {
out.nil_array();
return Ok(());
};
let element = args.get(2);
let Some(q) = body.c.get(element) else {
out.nil_array();
return Ok(());
};
let q = q.to_vec();
let near = body.c.search(&q, COUNT + 1, Some(element))?;
out.array(1);
write_hits(body, &q, &near, withscores, out);
Ok(())
}
fn vsetattr(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let Some(body) = write(db, args.get(1))? else {
out.bool(false);
return Ok(());
};
let element = args.get(2);
if !body.c.contains(element) {
out.bool(false);
return Ok(());
}
let value = args.get(3);
let side = body.side_mut(element);
side.attr = if value.is_empty() {
None
} else {
Some(value.into())
};
body.retag(element);
out.bool(true);
Ok(())
}
fn vgetattr(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
match read(db, args.get(1))?.and_then(|b| b.attr(args.get(2))) {
Some(attr) => out.bulk(attr),
None => out.nil(),
}
Ok(())
}
fn vrange(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let count = match args.len() {
4 => None,
5 => Some(args.get(4)),
_ => return Err(args::wrong_arity("VRANGE")),
};
let count = match count {
None => usize::MAX,
Some(arg) => match parse_i64(arg) {
Some(n) if n < 0 => usize::MAX,
Some(n) => usize::try_from(n).unwrap_or(usize::MAX),
None => return Err(Error::new(Code::Invalid, BAD_COUNT_VALUE)),
},
};
let start = bound(args.get(2), BAD_START)?;
let end = bound(args.get(3), BAD_END)?;
if matches!(start, Bound::Above) || matches!(end, Bound::Below) {
return Err(Error::new(Code::Invalid, BACKWARDS));
}
let Some(body) = read(db, args.get(1))? else {
out.array(0);
return Ok(());
};
let mut names: Vec<&[u8]> = (0..body.c.len())
.filter_map(|i| body.c.key_at(i))
.filter(|name| start.holds_start(name) && end.holds_end(name))
.collect();
names.sort_unstable();
names.truncate(count);
out.array(names.len());
for name in names {
out.bulk(name);
}
Ok(())
}
enum Bound<'a> {
Below,
Above,
In(&'a [u8]),
Out(&'a [u8]),
}
impl Bound<'_> {
fn holds_start(&self, name: &[u8]) -> bool {
match self {
Bound::Below => true,
Bound::Above => false,
Bound::In(at) => name >= *at,
Bound::Out(at) => name > *at,
}
}
fn holds_end(&self, name: &[u8]) -> bool {
match self {
Bound::Below => false,
Bound::Above => true,
Bound::In(at) => name <= *at,
Bound::Out(at) => name < *at,
}
}
}
fn bound<'a>(arg: &'a [u8], bad: &'static str) -> Result<Bound<'a>> {
match arg {
b"-" => Ok(Bound::Below),
b"+" => Ok(Bound::Above),
_ if arg.len() < 2 => Err(Error::new(Code::Invalid, bad)),
_ => match arg[0] {
b'[' => Ok(Bound::In(&arg[1..])),
b'(' => Ok(Bound::Out(&arg[1..])),
_ => Err(Error::new(Code::Invalid, bad)),
},
}
}
enum Query<'a> {
Element(&'a [u8]),
Vector(Vec<f32>),
}
struct Add<'a> {
asked: Requested,
quant: Quant,
attr: Option<&'a [u8]>,
}
impl<'a> Add<'a> {
fn parse(args: Args<'a>, from: usize) -> Result<Add<'a>> {
let mut got = Add {
asked: Requested::default(),
quant: Quant::Int8,
attr: None,
};
let mut i = from;
while i < args.len() {
let arg = args.get(i);
let rest = args.len() - i;
if args::is(arg, b"cas") {
i += 1;
} else if args::is(arg, b"noquant") {
got.quant = Quant::None;
i += 1;
} else if args::is(arg, b"bin") {
got.quant = Quant::Bin;
i += 1;
} else if args::is(arg, b"q8") {
got.quant = Quant::Int8;
i += 1;
} else if args::is(arg, b"ef") && rest >= 2 {
got.asked.ef_construction = positive(args.get(i + 1), BAD_EF)?;
i += 2;
} else if args::is(arg, b"m") && rest >= 2 {
got.asked.m = positive(args.get(i + 1), BAD_M)?;
i += 2;
} else if args::is(arg, b"setattr") && rest >= 2 {
got.attr = Some(args.get(i + 1));
i += 2;
} else {
return Err(args::syntax());
}
}
Ok(got)
}
}
struct Sim {
count: usize,
effort: usize,
withscores: bool,
withattribs: bool,
truth: bool,
filter: Option<vfilter::Filter>,
}
impl Sim {
fn parse(args: Args<'_>, from: usize) -> Result<Sim> {
let mut got = Sim {
count: COUNT,
effort: COUNT,
withscores: false,
withattribs: false,
truth: false,
filter: None,
};
let mut ef = None;
let mut i = from;
while i < args.len() {
let arg = args.get(i);
let rest = args.len() - i;
if args::is(arg, b"withscores") {
got.withscores = true;
i += 1;
} else if args::is(arg, b"withattribs") {
got.withattribs = true;
i += 1;
} else if args::is(arg, b"truth") {
got.truth = true;
i += 1;
} else if args::is(arg, b"nothread") {
i += 1;
} else if args::is(arg, b"count") && rest >= 2 {
got.count = positive(args.get(i + 1), BAD_COUNT)?;
i += 2;
} else if args::is(arg, b"ef") && rest >= 2 {
ef = Some(positive(args.get(i + 1), BAD_EF)?);
i += 2;
} else if args::is(arg, b"filter") && rest >= 2 {
got.filter = Some(vfilter::Filter::parse(args.get(i + 1))?);
i += 2;
} else if args::is(arg, b"filter-ef") && rest >= 2 {
let asked = match parse_i64(args.get(i + 1)) {
Some(n) => usize::try_from(n).unwrap_or(0),
None => return Err(Error::new(Code::Invalid, BAD_EF)),
};
ef = Some(ef.unwrap_or(0).max(asked));
i += 2;
} else {
return Err(args::syntax());
}
}
got.effort = got.count.max(ef.unwrap_or(0).min(1 << 16));
Ok(got)
}
}
fn search(
body: &VectorBody,
q: &[f32],
k: usize,
skip: Option<&[u8]>,
opts: &Sim,
) -> Result<Vec<Match>> {
let Some(expr) = &opts.filter else {
return if opts.truth {
body.c.search_exact(q, k, skip)
} else {
body.c.search(q, k, skip)
};
};
let want = Filtered {
expr,
want: expr.signature(),
side: &body.side,
};
if opts.truth {
body.c.search_exact_where(q, k, skip, &want)
} else {
body.c.search_where(q, k, skip, &want)
}
}
struct Filtered<'a> {
expr: &'a vfilter::Filter,
want: Signature,
side: &'a [Side],
}
impl yo_vector::Filter for Filtered<'_> {
fn allows(&self, tag: u64) -> bool {
Signature::from_bits(tag).covers(self.want)
}
fn exact(&self, id: u64) -> bool {
let attr = self.side.get(id as usize).and_then(|s| s.attr.as_deref());
self.expr.matches(attr)
}
}
fn answer(body: &VectorBody, q: &[f32], hits: &[Match], opts: &Sim, out: &mut Out) {
if !opts.withscores && !opts.withattribs {
out.array(hits.len());
for hit in hits {
out.bulk(&hit.key);
}
return;
}
let extras = usize::from(opts.withscores) + usize::from(opts.withattribs);
if out.proto().is_resp3() {
out.map(hits.len());
for hit in hits {
out.bulk(&hit.key);
if extras > 1 {
out.array(extras);
}
if opts.withscores {
out.double(score(body, q, hit));
}
if opts.withattribs {
attribute(body, &hit.key, out);
}
}
return;
}
out.array(hits.len() * (1 + extras));
for hit in hits {
out.bulk(&hit.key);
if opts.withscores {
out.double(score(body, q, hit));
}
if opts.withattribs {
attribute(body, &hit.key, out);
}
}
}
fn attribute(body: &VectorBody, key: &[u8], out: &mut Out) {
match body.attr(key) {
Some(attr) => out.bulk(attr),
None => out.nil(),
}
}
fn write_hits(body: &VectorBody, q: &[f32], hits: &[Match], withscores: bool, out: &mut Out) {
out.array(hits.len() * (1 + usize::from(withscores)));
for hit in hits {
out.bulk(&hit.key);
if withscores {
out.double(score(body, q, hit));
}
}
}
fn vector(args: Args<'_>, at: usize) -> Result<(Vec<f32>, usize)> {
let spec = args.opt(at).ok_or_else(args::syntax)?;
if args::is(spec, b"fp32") {
let blob = args.opt(at + 1).ok_or_else(args::syntax)?;
if blob.is_empty() || blob.len() % 4 != 0 {
return Err(Error::new(Code::Invalid, BAD_VECTOR));
}
let v = blob
.as_chunks::<4>()
.0
.iter()
.map(|b| f32::from_le_bytes(*b))
.collect();
return Ok((v, at + 2));
}
if args::is(spec, b"values") {
let n = positive(args.opt(at + 1).ok_or_else(args::syntax)?, BAD_VECTOR)?;
if args.len() < at + 2 + n {
return Err(args::syntax());
}
let mut v = Vec::with_capacity(n);
for j in 0..n {
let x = args.float(at + 2 + j)?;
#[allow(clippy::cast_possible_truncation)]
v.push(x as f32);
}
return Ok((v, at + 2 + n));
}
Err(Error::new(Code::Invalid, BAD_VECTOR))
}
fn draws(db: &mut Keyspace, count: i64, len: usize) -> Vec<usize> {
let Ok(want) = usize::try_from(count) else {
let repeats = usize::try_from(count.unsigned_abs()).unwrap_or(usize::MAX);
return (0..repeats.min(1 << 20))
.map(|_| (db.random() % len as u64) as usize)
.collect();
};
let mut all: Vec<usize> = (0..len).collect();
for i in (1..len).rev() {
all.swap(i, (db.random() % (i as u64 + 1)) as usize);
}
all.truncate(want);
all
}
fn open<'d>(
db: &'d mut Keyspace,
key: &[u8],
dim: usize,
quant: Quant,
asked: Requested,
) -> Result<&'d mut VectorBody> {
if db.kind_of(key).is_none() {
db.put_foreign(key, Box::new(VectorBody::new(dim, quant, asked)?));
}
match write(db, key)? {
Some(body) => {
if body.quant != quant {
return Err(Error::new(Code::Invalid, WRONG_QUANT));
}
if body.c.dim() != dim {
return Err(Error::fmt(
Code::Invalid,
format_args!(
"Vector dimension mismatch - got {dim} but set has {}",
body.c.dim()
),
));
}
Ok(body)
}
None => unreachable!("the vector set was just created"),
}
}
fn write<'d>(db: &'d mut Keyspace, key: &[u8]) -> Result<Option<&'d mut VectorBody>> {
match db.foreign_mut(key)? {
Some(body) => match body.downcast_mut::<VectorBody>() {
Some(body) => Ok(Some(body)),
None => Err(Error::new(Code::WrongType, NOT_A_VECTOR_SET)),
},
None => Ok(None),
}
}
fn read<'d>(db: &'d mut Keyspace, key: &[u8]) -> Result<Option<&'d VectorBody>> {
match db.foreign(key)? {
Some(body) => match body.downcast_ref::<VectorBody>() {
Some(body) => Ok(Some(body)),
None => Err(Error::new(Code::WrongType, NOT_A_VECTOR_SET)),
},
None => Ok(None),
}
}
fn positive(arg: &[u8], msg: &'static str) -> Result<usize> {
match parse_i64(arg) {
Some(n) if n > 0 => Ok(usize::try_from(n).unwrap_or(usize::MAX)),
_ => Err(Error::new(Code::Invalid, msg)),
}
}