Skip to content

fix(vector): persist database_id in vector index catalog entry #263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
EnRaiha wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(vector): persist database_id in vector index catalog entry #263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions nodedb-types/src/vector_index_params.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,12 @@
use serde::{Deserialize, Serialize};

/// Catalog entry for a vector index's build parameters.
///
/// NOTE: serialized as a zerompk ARRAY (the crate default — the
/// `default-as-map` feature is not enabled), so field ORDER is part of the
/// on-disk format. `database_id` was appended last; entries written by older
/// builds (11 fields, no database id) must still decode — see the legacy
/// fallback ladder in `catalog/vector_index_params.rs`.
#[derive(
Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
)]
Expand All @@ -36,6 +42,11 @@ pub struct StoredVectorIndexParams {
pub ivf_cells: usize,
/// IVF nprobe (0 = unused).
pub ivf_nprobe: usize,
/// Database that owns the collection. `0` = `DatabaseId::DEFAULT`,
/// which is also what entries written before this field existed decode
/// to (the legacy ladder fills 0) — matching the historical behavior of
/// the seed/rebuild paths.
pub database_id: u64,
}

#[cfg(test)]
Expand All @@ -56,11 +67,51 @@ mod tests {
pq_m: 0,
ivf_cells: 0,
ivf_nprobe: 0,
database_id: 7,
};
let bytes = zerompk::to_msgpack_vec(&e).unwrap();
let back: StoredVectorIndexParams = zerompk::from_msgpack(&bytes).unwrap();
assert_eq!(back.collection, "docs");
assert_eq!(back.dim, 4);
assert_eq!(back.field_name, "embedding");
assert_eq!(back.database_id, 7);
}

#[test]
fn legacy_11_field_array_decodes_with_database_id_zero() {
// Entries written before `database_id` existed are 11-element arrays
// in field order. The catalog module's decode ladder tolerates them
// by decoding the legacy tuple shape; here we verify the tuple shape
// itself round-trips so the ladder's fallback type is valid.
let legacy: (u64, String, String, usize, String, usize, usize, String, usize, usize, usize) =
(
1,
"docs".into(),
"embedding".into(),
4,
"cosine".into(),
16,
200,
String::new(),
0,
0,
0,
);
let bytes = zerompk::to_msgpack_vec(&legacy).unwrap();
let back: (
u64,
String,
String,
usize,
String,
usize,
usize,
String,
usize,
usize,
usize,
) = zerompk::from_msgpack(&bytes).unwrap();
assert_eq!(back.1, "docs");
assert_eq!(back.2, "embedding");
}
}
36 changes: 32 additions & 4 deletions nodedb/src/control/security/catalog/vector_index_params.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,36 @@ use redb::{ReadableDatabase, ReadableTable};

use super::types::{SystemCatalog, VECTOR_INDEX_PARAMS, catalog_err};

/// Decode a stored vector-index entry, tolerating the legacy 11-field shape
/// written before `database_id` existed (field order is the zerompk array
/// format, so `database_id` is simply appended last). Legacy entries decode
/// to `database_id = 0` (`DatabaseId::DEFAULT`) — the historical assumption
/// of the seed/rebuild paths.
fn decode_vector_index_params(bytes: &[u8]) -> crate::Result<StoredVectorIndexParams> {
if let Ok(e) = zerompk::from_msgpack::<StoredVectorIndexParams>(bytes) {
return Ok(e);
}
// Legacy 11-field tuple: (tenant, collection, field, dim, metric, m,
// ef_construction, index_type, pq_m, ivf_cells, ivf_nprobe).
let legacy: (u64, String, String, usize, String, usize, usize, String, usize, usize, usize) =
zerompk::from_msgpack(bytes)
.map_err(|e| catalog_err("deser legacy vector index params", e))?;
Ok(StoredVectorIndexParams {
tenant_id: legacy.0,
collection: legacy.1,
field_name: legacy.2,
dim: legacy.3,
metric: legacy.4,
m: legacy.5,
ef_construction: legacy.6,
index_type: legacy.7,
pq_m: legacy.8,
ivf_cells: legacy.9,
ivf_nprobe: legacy.10,
database_id: 0,
})
}

impl SystemCatalog {
/// Store vector index parameters for a collection/field.
pub fn put_vector_index_params(&self, entry: &StoredVectorIndexParams) -> crate::Result<()> {
Expand Down Expand Up @@ -49,8 +79,7 @@ impl SystemCatalog {

match table.get(key.as_str()) {
Ok(Some(value)) => {
let entry: StoredVectorIndexParams = zerompk::from_msgpack(value.value())
.map_err(|e| catalog_err("deser vector index params", e))?;
let entry = decode_vector_index_params(value.value())?;
Ok(Some(entry))
}
Ok(None) => Ok(None),
Expand Down Expand Up @@ -114,8 +143,7 @@ pub(super) fn list_all_vector_index_params_in(
.map_err(|e| catalog_err("range vector index params", e))?
{
let (_, value) = item.map_err(|e| catalog_err("read vector index params", e))?;
let entry: StoredVectorIndexParams = zerompk::from_msgpack(value.value())
.map_err(|e| catalog_err("deser vector index params", e))?;
let entry = decode_vector_index_params(value.value())?;
entries.push(entry);
}
Ok(entries)
Expand Down
1 change: 1 addition & 0 deletions nodedb/src/control/server/shared/ddl/neutral/dsl/vector_index.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,7 @@ pub async fn create_vector_index(
pq_m: params.pq_m,
ivf_cells: params.ivf_cells,
ivf_nprobe: params.ivf_nprobe,
database_id: database_id.as_u64(),
})
.map_err(|e| {
ddl_err(
Expand Down
42 changes: 31 additions & 11 deletions nodedb/src/data/executor/core_loop/vector_index_rebuild.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,20 +27,22 @@ impl CoreLoop {
entries: &[nodedb_types::StoredVectorIndexParams],
) {
use std::collections::HashSet;
let db = crate::types::DatabaseId::DEFAULT.as_u64();

// One scan per (tenant, collection): `apply_point_put_vector_indexes`
// One scan per (db, tenant, collection): `apply_point_put_vector_indexes`
// re-indexes ALL of the document's vector fields, so multiple field
// indexes on one collection need only a single scan.
let mut seen: HashSet<(u64, String)> = HashSet::new();
let mut targets: Vec<(u64, String)> = Vec::new();
// indexes on one collection need only a single scan. `database_id`
// comes from the durable entry (0 = DEFAULT for pre-existing entries,
// matching the historical behavior of this path).
let mut seen: HashSet<(u64, u64, String)> = HashSet::new();
let mut targets: Vec<(u64, u64, String)> = Vec::new();
for e in entries {
if seen.insert((e.tenant_id, e.collection.clone())) {
targets.push((e.tenant_id, e.collection.clone()));
let key = (e.database_id, e.tenant_id, e.collection.clone());
if seen.insert(key.clone()) {
targets.push(key);
}
}

for (tenant_id, collection) in targets {
for (db, tenant_id, collection) in targets {
// `entries` comes from the `CREATE VECTOR INDEX` param seed, so
// every target here is a classic collection with a vector index
// over a document field, and `apply_point_put_vector_indexes`
Expand Down Expand Up @@ -82,11 +84,29 @@ impl CoreLoop {
if let Some(surrogate) =
crate::engine::document::store::doc_id_to_surrogate(doc_id)
{
let normalized =
crate::data::executor::scan_normalize::sparse_body_to_msgpack(
let normalized: std::borrow::Cow<[u8]> = match &body_format {
crate::data::executor::sparse_body_format::SparseBodyFormat::Strict(schema) => {
match crate::data::executor::strict_format::binary_tuple_to_msgpack(value, schema) {
Some(mp) => std::borrow::Cow::Owned(mp),
None => {
tracing::warn!(
core = self.core_id,
%collection,
doc_id = %doc_id,
"strict BT decode failed in vector rebuild — doc will be unsearchable"
);
crate::data::executor::scan_normalize::sparse_body_to_msgpack(
value,
body_format.as_format_ref(),
)
}
}
}
_ => crate::data::executor::scan_normalize::sparse_body_to_msgpack(
value,
body_format.as_format_ref(),
);
),
};
docs.push((surrogate, normalized.into_owned()));
}
Ok(())
Expand Down
10 changes: 6 additions & 4 deletions nodedb/src/data/executor/core_loop/vector_index_seed.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,12 +30,13 @@ impl CoreLoop {
/// `execute_set_vector_params` keys a live `CREATE VECTOR INDEX`.
/// Called once at core startup, before the durable HNSW rebuild.
///
/// `CREATE VECTOR INDEX` computes its vshard with `DatabaseId::DEFAULT`
/// and the stored entry carries no database id, so the seed keys under
/// `DatabaseId::DEFAULT` to match.
/// `CREATE VECTOR INDEX` computes its vshard with the session's
/// `database_id`, and the stored entry now carries that id. Entries
/// written before the field existed decode to `0`
/// (`DatabaseId::DEFAULT`) — the historical assumption of this path.
pub fn seed_vector_index_params(&mut self, entries: &[nodedb_types::StoredVectorIndexParams]) {
for e in entries {
let db = crate::types::DatabaseId::DEFAULT.as_u64();
let db = e.database_id;
let key = CoreLoop::vector_index_key(db, e.tenant_id, &e.collection, &e.field_name);
let (params, config) = build_index_config_from_stored(e);
if e.dim > 0 {
Expand Down Expand Up @@ -128,6 +129,7 @@ mod tests {
pq_m: 0,
ivf_cells: 0,
ivf_nprobe: 0,
database_id: 0,
}
}

Expand Down
89 changes: 59 additions & 30 deletions nodedb/src/data/executor/handlers/point/apply_put/vector/put.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,58 @@ use crate::data::executor::core_loop::CoreLoop;

use super::types::{VectorFieldInsert, VectorIndexDelta, VectorIndexPutParams};

/// Extract a float vector from a decoded document value.
///
/// Accepts both native msgpack arrays (the forward write path) and JSON
/// strings (schemaless bodies transcoded from a columnar TEXT/JSON column,
/// e.g. `"[0.1, 0.2, ...]"` or a plain comma/whitespace-separated list) so
/// the durable-store rebuild path indexes the same vectors a live PUT would.
fn floats_from_value(v: &nodedb_types::Value) -> Option<Vec<f32>> {
match v {
nodedb_types::Value::Array(arr) => {
let floats: Vec<f32> = arr
.iter()
.filter_map(|v| match v {
nodedb_types::Value::Float(f) => Some(*f as f32),
nodedb_types::Value::Integer(i) => Some(*i as f32),
nodedb_types::Value::Decimal(d) => {
use rust_decimal::prelude::ToPrimitive;
d.to_f32()
}
nodedb_types::Value::String(s) => s.parse::<f32>().ok(),
_ => None,
})
.collect();
if floats.is_empty() {
None
} else {
Some(floats)
}
}
nodedb_types::Value::String(s) => {
// Try JSON array first, then comma/space-separated numbers.
if let Ok(vals) = serde_json::from_str::<Vec<f64>>(s) {
return Some(vals.into_iter().map(|f| f as f32).collect());
}
let trimmed = s.trim().trim_start_matches('[').trim_end_matches(']');
let mut floats: Vec<f32> = Vec::new();
for tok in trimmed.split([',', ' ', '\t', '\n']) {
let tok = tok.trim();
if tok.is_empty() {
continue;
}
floats.push(tok.parse::<f32>().ok()?);
}
if floats.is_empty() {
None
} else {
Some(floats)
}
}
_ => None,
}
}

impl CoreLoop {
/// HNSW vector indexing side-effect: index declared strict-schema
/// `Vector(dim)` columns, or (schemaless) fields matched by registered
Expand Down Expand Up @@ -54,20 +106,9 @@ impl CoreLoop {
&& let nodedb_types::Value::Object(ref obj) = ndb_val
{
for (field_name, dim) in &vector_fields {
if let Some(nodedb_types::Value::Array(arr)) = obj.get(field_name) {
let floats: Vec<f32> = arr
.iter()
.filter_map(|v| match v {
nodedb_types::Value::Float(f) => Some(*f as f32),
nodedb_types::Value::Integer(i) => Some(*i as f32),
nodedb_types::Value::Decimal(d) => {
use rust_decimal::prelude::ToPrimitive;
d.to_f32()
}
nodedb_types::Value::String(s) => s.parse::<f32>().ok(),
_ => None,
})
.collect();
if let Some(v) = obj.get(field_name)
&& let Some(floats) = floats_from_value(v)
{
let index_key =
Self::vector_index_key(database_id, tid, collection, field_name);
self.check_vector_width(&index_key, field_name, floats.len())?;
Expand Down Expand Up @@ -156,21 +197,10 @@ impl CoreLoop {
&& let nodedb_types::Value::Object(ref obj) = ndb_val
{
for (params_key, field_name) in &schemaless_keys {
if let Some(nodedb_types::Value::Array(arr)) = obj.get(field_name) {
let floats: Vec<f32> = arr
.iter()
.filter_map(|v| match v {
nodedb_types::Value::Float(f) => Some(*f as f32),
nodedb_types::Value::Integer(i) => Some(*i as f32),
nodedb_types::Value::Decimal(d) => {
use rust_decimal::prelude::ToPrimitive;
d.to_f32()
}
nodedb_types::Value::String(s) => s.parse::<f32>().ok(),
_ => None,
})
.collect();
if !floats.is_empty() {
if let Some(v) = obj.get(field_name)
&& let Some(floats) = floats_from_value(v)
&& !floats.is_empty()
{
let params = self
.vector_params
.get(params_key)
Expand Down Expand Up @@ -213,7 +243,6 @@ impl CoreLoop {
inserts.push(delta);
}
}
}
}
}
}
Expand Down
Loading