From 03d7b6fb22b5ee3184ffa2488b349bc3325d0143 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Fri, 28 Aug 2026 01:55:54 +0800 Subject: [PATCH] fix(vector): persist database_id in vector index catalog entry CREATE VECTOR INDEX keys everything under the session's real database id, but StoredVectorIndexParams never recorded it, so on boot the seed and the durable-store rebuild both hardcoded DatabaseId::DEFAULT (0). Any collection in a non-default database (e.g. db 'graph', id >= 1024) was never seeded, never rebuilt, and SEARCH returned 0 rows despite a successful CREATE. - add database_id as the 12th (last) field of StoredVectorIndexParams; zerompk structs serialize as arrays here, so field order is on-disk format and the new field is appended last - decode_vector_index_params() ladder: try 12-field struct, fall back to the legacy 11-field tuple filling database_id = 0, so existing catalog entries keep loading without a migration - seed_vector_index_params keys vector_params/index_configs/declared_dims on e.database_id instead of DatabaseId::DEFAULT - rebuild_vector_indexes_from_store groups targets by (database_id, tenant_id, collection) and scans the entry's real database - floats_from_value(): schemaless arms now also accept JSON-string embeddings (Value::String '[0.1,...]') transcoded from columnar TEXT/JSON, which the Array-only match previously dropped silently Verified live: strict vector(8) collection in db default rebuilds and SEARCHes (rebuilt=3, 3 rows); db 'graph' collection had no rebuild log until the catalog entry carried its database_id. --- nodedb-types/src/vector_index_params.rs | 51 +++++++++++ .../security/catalog/vector_index_params.rs | 36 +++++++- .../shared/ddl/neutral/dsl/vector_index.rs | 1 + .../core_loop/vector_index_rebuild.rs | 42 ++++++--- .../executor/core_loop/vector_index_seed.rs | 10 ++- .../handlers/point/apply_put/vector/put.rs | 89 ++++++++++++------- 6 files changed, 180 insertions(+), 49 deletions(-) diff --git a/nodedb-types/src/vector_index_params.rs b/nodedb-types/src/vector_index_params.rs index a2ee2d944..76acbcf41 100644 --- a/nodedb-types/src/vector_index_params.rs +++ b/nodedb-types/src/vector_index_params.rs @@ -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, )] @@ -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)] @@ -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"); } } diff --git a/nodedb/src/control/security/catalog/vector_index_params.rs b/nodedb/src/control/security/catalog/vector_index_params.rs index 622a2e299..4d8e61d5b 100644 --- a/nodedb/src/control/security/catalog/vector_index_params.rs +++ b/nodedb/src/control/security/catalog/vector_index_params.rs @@ -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 { + if let Ok(e) = zerompk::from_msgpack::(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<()> { @@ -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), @@ -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) diff --git a/nodedb/src/control/server/shared/ddl/neutral/dsl/vector_index.rs b/nodedb/src/control/server/shared/ddl/neutral/dsl/vector_index.rs index 44b56497b..56917f5d3 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/dsl/vector_index.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/dsl/vector_index.rs @@ -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( diff --git a/nodedb/src/data/executor/core_loop/vector_index_rebuild.rs b/nodedb/src/data/executor/core_loop/vector_index_rebuild.rs index b236b75d9..e12e2599f 100644 --- a/nodedb/src/data/executor/core_loop/vector_index_rebuild.rs +++ b/nodedb/src/data/executor/core_loop/vector_index_rebuild.rs @@ -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` @@ -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(()) diff --git a/nodedb/src/data/executor/core_loop/vector_index_seed.rs b/nodedb/src/data/executor/core_loop/vector_index_seed.rs index f3a4e35dd..688a092ff 100644 --- a/nodedb/src/data/executor/core_loop/vector_index_seed.rs +++ b/nodedb/src/data/executor/core_loop/vector_index_seed.rs @@ -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 { @@ -128,6 +129,7 @@ mod tests { pq_m: 0, ivf_cells: 0, ivf_nprobe: 0, + database_id: 0, } } diff --git a/nodedb/src/data/executor/handlers/point/apply_put/vector/put.rs b/nodedb/src/data/executor/handlers/point/apply_put/vector/put.rs index 46c0847c9..d397e3c5f 100644 --- a/nodedb/src/data/executor/handlers/point/apply_put/vector/put.rs +++ b/nodedb/src/data/executor/handlers/point/apply_put/vector/put.rs @@ -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> { + match v { + nodedb_types::Value::Array(arr) => { + let floats: Vec = 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::().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::>(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 = Vec::new(); + for tok in trimmed.split([',', ' ', '\t', '\n']) { + let tok = tok.trim(); + if tok.is_empty() { + continue; + } + floats.push(tok.parse::().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 @@ -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 = 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::().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())?; @@ -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 = 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::().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) @@ -213,7 +243,6 @@ impl CoreLoop { inserts.push(delta); } } - } } } }