diff --git a/lib/edge/python/qdrant_edge.pyi b/lib/edge/python/qdrant_edge.pyi index 82f9641aa5..36b438a9ee 100644 --- a/lib/edge/python/qdrant_edge.pyi +++ b/lib/edge/python/qdrant_edge.pyi @@ -254,7 +254,7 @@ class EdgeConfig: Union["EdgeVectorParams", Dict[str, "EdgeVectorParams"]] ] = None, sparse_vectors: Optional[Dict[str, "EdgeSparseVectorParams"]] = None, - on_disk_payload: bool = True, + on_disk_payload: Optional[bool] = None, hnsw_config: Optional["HnswIndexConfig"] = None, quantization_config: Optional[QuantizationConfigType] = None, optimizers: Optional["EdgeOptimizersConfig"] = None, @@ -263,12 +263,19 @@ class EdgeConfig: """ Create an EdgeConfig. + Parameters left as None are "not specified": when loading an existing shard each + one resolves through provided -> persisted -> derived from segments -> default, + so an unspecified parameter keeps the shard as it is. vectors and sparse_vectors + define the stored data: if provided they are validated for compatibility against + the existing segments, if omitted they are inherited from the shard. + Args: vectors: Dense vector configuration. Can be a single EdgeVectorParams for the default vector (name "") or a dict of name -> EdgeVectorParams. Optional if sparse_vectors is provided (sparse-only config). sparse_vectors: Optional sparse vector configurations. on_disk_payload: If True, store payload on disk (mmap); otherwise in RAM. + None keeps the shard's current value (defaults to on-disk). hnsw_config: Optional global HNSW config (used when building HNSW index). quantization_config: Optional global quantization config. optimizers: Optional optimizer settings. @@ -290,13 +297,13 @@ class EdgeConfig: ... @property - def on_disk_payload(self) -> bool: - """Whether payload is stored on disk.""" + def on_disk_payload(self) -> Optional[bool]: + """Whether payload is stored on disk, or None if not specified.""" ... @property - def hnsw_config(self) -> "HnswIndexConfig": - """Global HNSW config.""" + def hnsw_config(self) -> Optional["HnswIndexConfig"]: + """Global HNSW config, or None if not specified.""" ... @property @@ -305,8 +312,8 @@ class EdgeConfig: ... @property - def optimizers(self) -> "EdgeOptimizersConfig": - """Optimizer settings.""" + def optimizers(self) -> Optional["EdgeOptimizersConfig"]: + """Optimizer settings, or None if not specified.""" ... @property diff --git a/lib/edge/python/src/config/mod.rs b/lib/edge/python/src/config/mod.rs index 5e95f53caa..db0629a8e8 100644 --- a/lib/edge/python/src/config/mod.rs +++ b/lib/edge/python/src/config/mod.rs @@ -25,13 +25,13 @@ pub struct PyEdgeConfig(pub EdgeConfig); #[pymethods] impl PyEdgeConfig { #[new] - #[pyo3(signature = (vectors=None, sparse_vectors=None, on_disk_payload=true, hnsw_config=None, quantization_config=None, optimizers=None, max_search_threads=None))] + #[pyo3(signature = (vectors=None, sparse_vectors=None, on_disk_payload=None, hnsw_config=None, quantization_config=None, optimizers=None, max_search_threads=None))] pub fn new( #[pyo3(from_py_with = option_edge_vectors_helper)] vectors: Option< HashMap, >, sparse_vectors: Option>, - on_disk_payload: bool, + on_disk_payload: Option, hnsw_config: Option, quantization_config: Option, optimizers: Option, @@ -52,9 +52,9 @@ impl PyEdgeConfig { on_disk_payload, vectors, sparse_vectors, - hnsw_config: hnsw_config.map(|h| h.0).unwrap_or_default(), + hnsw_config: hnsw_config.map(|h| h.0), quantization_config: quantization_config.map(QuantizationConfig::from), - optimizers: optimizers.map(|o| o.0).unwrap_or_default(), + optimizers: optimizers.map(|o| o.0), wal_options: None, max_search_threads, })) @@ -71,13 +71,13 @@ impl PyEdgeConfig { } #[getter] - pub fn on_disk_payload(&self) -> bool { + pub fn on_disk_payload(&self) -> Option { self.0.on_disk_payload } #[getter] - pub fn hnsw_config(&self) -> PyHnswIndexConfig { - PyHnswIndexConfig(self.0.hnsw_config) + pub fn hnsw_config(&self) -> Option { + self.0.hnsw_config.map(PyHnswIndexConfig) } #[getter] @@ -86,8 +86,8 @@ impl PyEdgeConfig { } #[getter] - pub fn optimizers(&self) -> PyEdgeOptimizersConfig { - PyEdgeOptimizersConfig(self.0.optimizers.clone()) + pub fn optimizers(&self) -> Option { + self.0.optimizers.clone().map(PyEdgeOptimizersConfig) } #[getter] diff --git a/lib/edge/src/builders/edge_config.rs b/lib/edge/src/builders/edge_config.rs index f24ab1dff5..3d988e2730 100644 --- a/lib/edge/src/builders/edge_config.rs +++ b/lib/edge/src/builders/edge_config.rs @@ -15,9 +15,10 @@ use crate::config::vectors::{EdgeSparseVectorParams, EdgeVectorParams}; /// Fluent builder for [`EdgeConfig`]. /// -/// All fields are optional and fall back to [`EdgeConfig::default`] values -/// at [`Self::build`] time; at minimum supply at least one dense or sparse -/// vector via [`Self::vector`] / [`Self::sparse_vector`]. +/// All fields are optional; at minimum supply at least one dense or sparse +/// vector via [`Self::vector`] / [`Self::sparse_vector`]. Fields left unset +/// stay unspecified (`None`) in the built config: loading an existing shard +/// keeps their persisted values, otherwise defaults apply. #[derive(Debug, Default)] pub struct EdgeConfigBuilder { on_disk_payload: Option, @@ -113,16 +114,15 @@ impl EdgeConfigBuilder { wal_options, max_search_threads, } = self; - let defaults = EdgeConfig::default(); EdgeConfig { - on_disk_payload: on_disk_payload.unwrap_or(defaults.on_disk_payload), + on_disk_payload, vectors, sparse_vectors, - hnsw_config: hnsw_config.unwrap_or(defaults.hnsw_config), - quantization_config: quantization_config.or(defaults.quantization_config), - optimizers: optimizers.unwrap_or(defaults.optimizers), - wal_options: wal_options.or(defaults.wal_options), - max_search_threads: max_search_threads.or(defaults.max_search_threads), + hnsw_config, + quantization_config, + optimizers, + wal_options, + max_search_threads, } } } diff --git a/lib/edge/src/config/shard.rs b/lib/edge/src/config/shard.rs index e1cc8be27f..6182571507 100644 --- a/lib/edge/src/config/shard.rs +++ b/lib/edge/src/config/shard.rs @@ -19,27 +19,40 @@ use super::vectors::{EdgeSparseVectorParams, EdgeVectorParams}; pub(crate) const EDGE_CONFIG_FILE: &str = "edge_config.json"; /// Full configuration for an edge shard. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +/// +/// `vectors` and `sparse_vectors` define the stored data: when loading an existing shard they are +/// validated for compatibility against the segments if provided (non-empty), or taken from the +/// persisted config / the segments themselves if not. +/// +/// Everything else is tunable and `None` means "not specified": when loading an existing shard +/// each parameter resolves through provided → persisted → derived from segments → default (see +/// [`EdgeConfig::fill_unspecified_from`]), so leaving a parameter unspecified keeps the shard as +/// it is, while a `Some` value explicitly overwrites it and existing segments converge to it +/// through the optimizers. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub struct EdgeConfig { /// If true, payload is stored on disk (mmap); otherwise in RAM. Same as `CollectionParams::on_disk_payload`. - #[serde(default = "default_on_disk_payload")] - pub on_disk_payload: bool, + /// `None` defaults to on-disk, see [`EdgeConfig::on_disk_payload`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub on_disk_payload: Option, /// Dense vector params per vector name. #[serde(default)] pub vectors: HashMap, /// Sparse vector params per vector name. #[serde(default)] pub sparse_vectors: HashMap, - /// Global HNSW config; per-vector override is in `vectors[].hnsw_config` - #[serde(default)] - pub hnsw_config: HnswConfig, + /// Global HNSW config; per-vector override is in `vectors[].hnsw_config`. + /// `None` defaults to [`HnswConfig::default`], see [`EdgeConfig::hnsw_config`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub hnsw_config: Option, /// Global quantization config for all vectors /// Per-vector override in in `vectors[].quantization_config` #[serde(default, skip_serializing_if = "Option::is_none")] pub quantization_config: Option, - #[serde(default)] - pub optimizers: EdgeOptimizersConfig, + /// `None` defaults to [`EdgeOptimizersConfig::default`], see [`EdgeConfig::optimizers`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub optimizers: Option, /// WAL options for the shard. `None` keeps the WAL crate's defaults /// (32 MiB segment capacity). Override for embedded/mobile deployments /// where the default segment size is too large. @@ -53,31 +66,82 @@ pub struct EdgeConfig { pub max_search_threads: Option, } -fn default_on_disk_payload() -> bool { - true -} - -impl Default for EdgeConfig { - fn default() -> Self { - Self { - on_disk_payload: default_on_disk_payload(), - vectors: HashMap::new(), - sparse_vectors: HashMap::new(), - hnsw_config: HnswConfig::default(), - quantization_config: None, - optimizers: EdgeOptimizersConfig::default(), - wal_options: None, - max_search_threads: None, - } - } -} - impl EdgeConfig { /// Start building an [`EdgeConfig`] with a fluent API. pub fn builder() -> crate::builders::EdgeConfigBuilder { crate::builders::EdgeConfigBuilder::new() } + /// Effective payload storage location: on-disk unless explicitly set to `false`. + pub fn on_disk_payload(&self) -> bool { + self.on_disk_payload.unwrap_or(true) + } + + /// Effective global HNSW config: [`HnswConfig::default`] unless explicitly set. + pub fn hnsw_config(&self) -> HnswConfig { + self.hnsw_config.unwrap_or_default() + } + + /// Effective optimizers config: [`EdgeOptimizersConfig::default`] unless explicitly set. + pub fn optimizers(&self) -> EdgeOptimizersConfig { + self.optimizers.clone().unwrap_or_default() + } + + /// Fill parameters left unspecified from `base`, keeping explicitly provided values. + /// + /// Chained over the fallback layers of [`EdgeShard::load`](crate::EdgeShard::load): + /// provided → persisted → derived from segments → default. + /// + /// For tunables, unspecified means `None`. For `vectors` and `sparse_vectors` it means an + /// empty map: a non-empty map is taken as-is (never merged element-wise) — those define the + /// stored data, so the load path validates them against existing segments instead of + /// converging via the optimizers like the tunables do. + pub fn fill_unspecified_from(self, base: &EdgeConfig) -> Self { + let Self { + on_disk_payload, + vectors, + sparse_vectors, + hnsw_config, + quantization_config, + optimizers, + wal_options, + max_search_threads, + } = self; + Self { + on_disk_payload: on_disk_payload.or(base.on_disk_payload), + vectors: if vectors.is_empty() { + base.vectors.clone() + } else { + vectors + }, + sparse_vectors: if sparse_vectors.is_empty() { + base.sparse_vectors.clone() + } else { + sparse_vectors + }, + hnsw_config: hnsw_config.or(base.hnsw_config), + quantization_config: quantization_config.or_else(|| base.quantization_config.clone()), + optimizers: optimizers.or_else(|| base.optimizers.clone()), + wal_options: wal_options.or_else(|| base.wal_options.clone()), + max_search_threads: max_search_threads.or(base.max_search_threads), + } + } + + /// Accumulate the config derived from one more segment into `acc`. + /// + /// Building block for the "derived from segments" layer of the config fallback chain: fold + /// this over *all* segments, so that a segment carrying no information about a parameter + /// (e.g. a plain appendable segment says nothing about HNSW) never masks one that does (an + /// indexed segment carries the actual build parameters). Fold in a deterministic segment + /// order: when segments disagree on a parameter, the first one providing it wins. + pub(crate) fn fold_from_segment_config(acc: Option, segment: &SegmentConfig) -> Self { + let derived = Self::from_segment_config(segment); + match acc { + Some(acc) => acc.fill_unspecified_from(&derived), + None => derived, + } + } + /// Build from existing segment config. Fills all parameters that can be inferred. pub fn from_segment_config(segment: &SegmentConfig) -> Self { let SegmentConfig { @@ -111,24 +175,21 @@ impl EdgeConfig { segment::types::Indexes::Hnsw(h) => Some(*h), }) .collect(); - let hnsw_config = hnsw_configs - .first() - .and_then(|first| { - if hnsw_configs.iter().all(|h| h == first) { - Some(*first) - } else { - None - } - }) - .unwrap_or_default(); + let hnsw_config = hnsw_configs.first().and_then(|first| { + if hnsw_configs.iter().all(|h| h == first) { + Some(*first) + } else { + None + } + }); Self { - on_disk_payload, + on_disk_payload: Some(on_disk_payload), vectors, sparse_vectors, hnsw_config, quantization_config: None, - optimizers: EdgeOptimizersConfig::default(), + optimizers: None, wal_options: None, max_search_threads: None, } @@ -152,7 +213,7 @@ impl EdgeConfig { /// Segment config for creating appendable segments only. /// Does not contain any HNSW configuration (plain index only). pub fn plain_segment_config(&self) -> SegmentConfig { - let payload_storage_type = PayloadStorageType::from_on_disk_payload(self.on_disk_payload); + let payload_storage_type = PayloadStorageType::from_on_disk_payload(self.on_disk_payload()); let vector_data = self .vectors .iter() @@ -200,6 +261,7 @@ impl EdgeConfig { payload_storage_type, } = self.plain_segment_config(); + let hnsw_config = self.hnsw_config(); let dense_vector = self .vectors .iter() @@ -207,7 +269,7 @@ impl EdgeConfig { ( name.clone(), p.to_dense_vector_optimizer_config( - &self.hnsw_config, + &hnsw_config, self.quantization_config.as_ref(), ), ) @@ -242,13 +304,11 @@ impl EdgeConfig { } pub fn optimizer_thresholds(&self, num_indexing_threads: usize) -> OptimizerThresholds { - let indexing_threshold_kb = self.optimizers.get_indexing_threshold_kb(); + let optimizers = self.optimizers(); OptimizerThresholds { memmap_threshold_kb: usize::MAX, - indexing_threshold_kb, - max_segment_size_kb: self - .optimizers - .get_max_segment_size_kb(num_indexing_threads), + indexing_threshold_kb: optimizers.get_indexing_threshold_kb(), + max_segment_size_kb: optimizers.get_max_segment_size_kb(num_indexing_threads), deferred_internal_id: None, } } @@ -275,7 +335,7 @@ impl EdgeConfig { } pub fn set_hnsw_config(&mut self, hnsw_config: HnswConfig) { - self.hnsw_config = hnsw_config; + self.hnsw_config = Some(hnsw_config); } pub fn set_vector_hnsw_config( @@ -293,6 +353,55 @@ impl EdgeConfig { } pub fn set_optimizers_config(&mut self, optimizers: EdgeOptimizersConfig) { - self.optimizers = optimizers; + self.optimizers = Some(optimizers); + } +} + +#[cfg(test)] +mod tests { + use segment::types::{Distance, Indexes, VectorDataConfig, VectorStorageType}; + + use super::*; + + fn segment_config(index: Indexes) -> SegmentConfig { + SegmentConfig { + vector_data: HashMap::from([( + "vec".to_string(), + VectorDataConfig { + size: 4, + distance: Distance::Dot, + storage_type: VectorStorageType::ChunkedMmap, + index, + quantization_config: None, + multivector_config: None, + datatype: None, + }, + )]), + sparse_vector_data: HashMap::new(), + payload_storage_type: PayloadStorageType::from_on_disk_payload(true), + } + } + + /// A plain (appendable) segment carries no HNSW parameters; folding must not let it mask an + /// indexed segment's actual build parameters, regardless of segment order. + #[test] + fn fold_derives_hnsw_from_indexed_segment_regardless_of_order() { + let hnsw = HnswConfig { + m: 32, + ..HnswConfig::default() + }; + let plain = segment_config(Indexes::Plain {}); + let indexed = segment_config(Indexes::Hnsw(hnsw)); + + for segments in [[&plain, &indexed], [&indexed, &plain]] { + let derived = segments + .into_iter() + .fold(None, |acc, segment| { + Some(EdgeConfig::fold_from_segment_config(acc, segment)) + }) + .unwrap(); + assert_eq!(derived.hnsw_config, Some(hnsw)); + assert!(derived.vectors.contains_key("vec")); + } } } diff --git a/lib/edge/src/lib.rs b/lib/edge/src/lib.rs index 4360450e43..6f9deb7f25 100644 --- a/lib/edge/src/lib.rs +++ b/lib/edge/src/lib.rs @@ -110,10 +110,15 @@ impl EdgeShard { /// Load an edge shard from existing files at `path`. /// - /// * If `config` is `Some`: check compatibility with loaded segments, then overwrite - /// `edge_config.json` with it. - /// * If `config` is `None`: load config from `edge_config.json`, or infer from segments; - /// check compatibility, then persist so future loads have it. + /// Every tunable parameter resolves through the fallback chain + /// **provided → persisted (`edge_config.json`) → derived from segments → default**, so a + /// parameter left unspecified (`None`) keeps whatever the shard already has, while an + /// explicitly provided value overwrites it and existing segments converge to it through the + /// optimizers. The resolved config is persisted to `edge_config.json`. + /// + /// `vectors` and `sparse_vectors` define the stored data and cannot be changed here: if + /// provided (non-empty), they are validated for compatibility against the loaded segments; + /// if not, they are taken from the persisted config or the segments themselves. /// /// Fails if no segments exist and no config can be loaded or inferred. /// @@ -121,30 +126,40 @@ impl EdgeShard { /// the default 32 MiB segment capacity is too large), set /// [`EdgeConfig::wal_options`] on the supplied config. pub fn load(path: &Path, config: Option) -> OperationResult { - let mut config = resolve_initial_config(path, config)?; + let resolved = resolve_initial_config(path, config)?; - let wal_options = config + let wal_options = resolved .as_ref() .and_then(|c| c.wal_options.clone()) .unwrap_or_default(); let (wal, segments_path) = ensure_dirs_and_open_wal(path, wal_options)?; - let mut segments = load_segments(path, &segments_path, &mut config)?; + let (mut segments, derived) = load_segments(&segments_path)?; - ensure_appendable_segment( - &mut segments, - path, - &segments_path, - config.as_ref().ok_or_else(|| { - OperationError::service_error( + let config = match (resolved, derived) { + (Some(resolved), Some(derived)) => { + let merged = resolved.fill_unspecified_from(&derived); + // The tunables converge to the merged config via the optimizers, but the vector + // definitions must actually match the stored data. + merged + .check_compatible_with_segment_config(&derived.plain_segment_config()) + .map_err(|err| { + OperationError::service_error(format!( + "config is incompatible with existing segments: {err}" + )) + })?; + merged + } + (Some(resolved), None) => resolved, + (None, Some(derived)) => derived, + (None, None) => { + return Err(OperationError::service_error( "edge config is not provided and no segments were loaded", - ) - })?, - )?; + )); + } + }; - let config = config.ok_or_else(|| { - OperationError::service_error("edge config is not provided and no segments were loaded") - })?; + ensure_appendable_segment(&mut segments, path, &segments_path, &config)?; let search_pool = pool::build_search_pool(config.search_thread_count())?; @@ -305,17 +320,22 @@ fn ensure_dirs_and_open_wal( Ok((wal, segments_path)) } +/// The provided → persisted layers of the config fallback chain (the derived-from-segments layer +/// is applied by [`EdgeShard::load`] once the segments are loaded). fn resolve_initial_config( path: &Path, config: Option, ) -> OperationResult> { - Ok(match config { - Some(c) => Some(c), - None => match EdgeConfig::load(path) { - Some(Ok(c)) => Some(c), - Some(Err(e)) => return Err(e), - None => None, - }, + let persisted = match EdgeConfig::load(path) { + Some(Ok(c)) => Some(c), + Some(Err(e)) => return Err(e), + None => None, + }; + Ok(match (config, persisted) { + // Provided config wins, but parameters it leaves unspecified keep their persisted values + (Some(provided), Some(persisted)) => Some(provided.fill_unspecified_from(&persisted)), + (Some(provided), None) => Some(provided), + (None, persisted) => persisted, }) } @@ -370,14 +390,18 @@ pub(crate) fn scan_segment_dirs(segments_path: &Path) -> OperationResult, -) -> OperationResult { +/// Load all segments and fold their configs into a single derived [`EdgeConfig`] — the +/// derived-from-segments layer of the config fallback chain. Segments are folded in UUID order so +/// the derivation is deterministic, and each segment is checked for compatibility against the +/// previously loaded ones. +fn load_segments(segments_path: &Path) -> OperationResult<(SegmentHolder, Option)> { let mut segments = SegmentHolder::default(); + let mut derived: Option = None; - for (segment_uuid, segment_path) in scan_segment_dirs(segments_path)? { + let mut segment_dirs: Vec<_> = scan_segment_dirs(segments_path)?.into_iter().collect(); + segment_dirs.sort_unstable_by_key(|(segment_uuid, _)| *segment_uuid); + + for (segment_uuid, segment_path) in segment_dirs { let mut segment = load_segment(&segment_path, segment_uuid, None, &AtomicBool::new(false)) .map_err(|err| { OperationError::service_error(format!( @@ -387,16 +411,16 @@ fn load_segments( })?; let segment_cfg = segment.config(); - if let Some(cfg) = config.as_ref() { - cfg.check_compatible_with_segment_config(segment_cfg).map_err( - |err| OperationError::service_error(format!( - "segment {} is incompatible with provided config or previously loaded segments: {err}", - segment_path.display(), - )) - )?; - } else { - *config = Some(EdgeConfig::from_segment_config(segment_cfg)); + if let Some(acc) = derived.as_ref() { + acc.check_compatible_with_segment_config(segment_cfg) + .map_err(|err| { + OperationError::service_error(format!( + "segment {} is incompatible with previously loaded segments: {err}", + segment_path.display(), + )) + })?; } + derived = Some(EdgeConfig::fold_from_segment_config(derived, segment_cfg)); segment.check_consistency_and_repair().map_err(|err| { OperationError::service_error(format!( @@ -408,7 +432,7 @@ fn load_segments( segments.add_new(segment); } - Ok(segments) + Ok((segments, derived)) } fn ensure_appendable_segment( diff --git a/lib/edge/src/optimize.rs b/lib/edge/src/optimize.rs index 07fa4ec99b..7aa52c67b3 100644 --- a/lib/edge/src/optimize.rs +++ b/lib/edge/src/optimize.rs @@ -100,11 +100,12 @@ impl EdgeShard { let segment_optimizer_config = cfg .segment_optimizer_config() .with_live_vector_names(live_vector_names); - let global_hnsw_config = cfg.hnsw_config; + let global_hnsw_config = cfg.hnsw_config(); + let optimizers_config = cfg.optimizers(); let hnsw_global_config = HnswGlobalConfig::default(); let num_indexing_threads = max_num_indexing_threads(&segment_optimizer_config); let threshold_config = cfg.optimizer_thresholds(num_indexing_threads); - let default_segments_number = cfg.optimizers.get_number_segments(); + let default_segments_number = optimizers_config.get_number_segments(); vec![ Arc::new(MergeOptimizer::new( @@ -124,10 +125,10 @@ impl EdgeShard { hnsw_global_config.clone(), )), Arc::new(VacuumOptimizer::new( - cfg.optimizers + optimizers_config .deleted_threshold .unwrap_or(DEFAULT_DELETED_THRESHOLD), - cfg.optimizers + optimizers_config .vacuum_min_vector_number .unwrap_or(DEFAULT_VACUUM_MIN_VECTOR_NUMBER), threshold_config, @@ -867,7 +868,7 @@ mod tests { fn test_config() -> EdgeConfig { EdgeConfig { - on_disk_payload: false, + on_disk_payload: Some(false), vectors: HashMap::from([( VECTOR_NAME.to_string(), EdgeVectorParams { @@ -881,9 +882,9 @@ mod tests { }, )]), sparse_vectors: HashMap::new(), - hnsw_config: Default::default(), + hnsw_config: None, quantization_config: None, - optimizers: Default::default(), + optimizers: None, wal_options: None, max_search_threads: None, } diff --git a/lib/edge/src/read_only/lifecycle.rs b/lib/edge/src/read_only/lifecycle.rs index b92c8a1f63..4351adf507 100644 --- a/lib/edge/src/read_only/lifecycle.rs +++ b/lib/edge/src/read_only/lifecycle.rs @@ -17,10 +17,26 @@ impl ReadOnlyEdgeShard { /// leader's segment manifest. Requires the leader to write a manifest (the /// `write_segment_manifest` feature flag). pub fn open_mmap(path: &Path) -> OperationResult { - Self::open(MmapFs, path) + Self::open(MmapFs, path, None) } } +/// Effective follower config: tunables prefer the caller-`provided` config and fall back to the +/// segment-`derived` one (via [`EdgeConfig::fill_unspecified_from`]), while `vectors` and +/// `sparse_vectors` always come from `derived` — the segments are the follower's source of truth +/// for the stored data. +pub(super) fn merge_follower_config(provided: EdgeConfig, mut derived: EdgeConfig) -> EdgeConfig { + // Take the vector params out of `derived` up front, so caller-provided ones cannot win in + // the fill below (a non-empty map would count as "specified"). + let vectors = std::mem::take(&mut derived.vectors); + let sparse_vectors = std::mem::take(&mut derived.sparse_vectors); + + let mut config = provided.fill_unspecified_from(&derived); + config.vectors = vectors; + config.sparse_vectors = sparse_vectors; + config +} + impl ReadOnlyEdgeShard { /// Open a read-only follower over the edge-shard directory at `path` using read backend `fs`. /// @@ -29,14 +45,16 @@ impl ReadOnlyEdgeShard { /// /// A follower has no `edge_config.json` — the segments are the source of truth, so the config is /// derived from the segments themselves (see [`EdgeConfig::from_segment_config`]), mirroring the - /// read-write [`EdgeShard`](crate::EdgeShard)'s fallback. An empty shard (no segments yet) starts - /// from a default config and re-derives one once segments appear on [`refresh`](Self::refresh). - pub fn open(fs: S::Fs, path: &Path) -> OperationResult + /// read-write [`EdgeShard`](crate::EdgeShard)'s fallback. A provided `config` overrides tunable + /// parameters at open (its `Some` values win over the derived ones; `vectors`/`sparse_vectors` + /// are ignored); a [`refresh`](Self::refresh) re-derives the config from the segments alone. + /// An empty shard (no segments yet) starts from the provided config (or a default one). + pub fn open(fs: S::Fs, path: &Path, config: Option) -> OperationResult where S::Fs: Send + Sync + Clone + 'static, { let enumerator = ManifestSegmentEnumerator::new(fs.clone(), path); - Self::open_with_enumerator(fs, path, enumerator) + Self::open_with_enumerator(fs, path, enumerator, config) } /// Open with an explicit segment [`enumerator`](SegmentEnumerator). @@ -50,30 +68,45 @@ impl ReadOnlyEdgeShard { fs: S::Fs, path: &Path, enumerator: impl SegmentEnumerator + 'static, + config: Option, ) -> OperationResult where S::Fs: Send + Sync + Clone + 'static, { - // A follower has no `edge_config.json` and derives its config from the segments — which never - // carry `max_search_threads` — so the pool is always sized from the CPU-derived default. - let search_pool = - crate::pool::build_search_pool(EdgeConfig::default().search_thread_count())?; + let provided_config = config.unwrap_or_default(); - let loaded = load_segments_parallel::(&search_pool, &fs, enumerator.list_segments()?)?; + // Segments never carry `max_search_threads`, so the pool is sized from the caller-provided + // config alone: the CPU-derived default unless explicitly set. + let search_pool = crate::pool::build_search_pool(provided_config.search_thread_count())?; + let mut loaded = + load_segments_parallel::(&search_pool, &fs, enumerator.list_segments()?)?; + // Fold the derived config in UUID order so the derivation is deterministic. + loaded.sort_unstable_by_key(|(uuid, _)| *uuid); let mut holder = ReadOnlySegmentHolder::default(); - let mut config: Option = None; + let mut segments_config: Option = None; for (uuid, segment) in loaded { - // Derive the shard config from the first segment's own config. - config.get_or_insert_with(|| EdgeConfig::from_segment_config(&segment.segment_config)); + // Derive the shard config from the segments' own configs: folded over all of them, so + // a segment carrying no information about a parameter (e.g. a plain appendable one + // says nothing about HNSW) never masks one that does. + segments_config = Some(EdgeConfig::fold_from_segment_config( + segments_config, + &segment.segment_config, + )); let appendable = segment.segment_config.is_appendable(); holder.insert(uuid, appendable, Arc::new(RwLock::new(segment))); } + let config = match segments_config { + Some(derived) => merge_follower_config(provided_config, derived), + // Empty shard: no segments to derive from yet, refresh re-derives once they appear. + None => provided_config, + }; + Ok(Self { path: path.to_path_buf(), fs, - config: RwLock::new(Arc::new(config.unwrap_or_default())), + config: RwLock::new(Arc::new(config)), segments: RwLock::new(holder), enumerator: Box::new(enumerator), search_pool, diff --git a/lib/edge/src/read_only/mod.rs b/lib/edge/src/read_only/mod.rs index 6c314d9177..e8bae192d5 100644 --- a/lib/edge/src/read_only/mod.rs +++ b/lib/edge/src/read_only/mod.rs @@ -43,8 +43,9 @@ pub struct ReadOnlyEdgeShard { path: PathBuf, /// Read backend handle; passed to segment `open` and `live_reload`. fs: S::Fs, - /// Config snapshot, derived from the segments (a follower has no `edge_config.json`) and - /// re-derived on each refresh. Stored as an `Arc` so a read view can cheaply clone the current + /// Config snapshot, derived from the segments (a follower has no `edge_config.json`). At open + /// it is overlaid with the tunables of the caller-provided config; each refresh re-derives it + /// from the segments alone. Stored as an `Arc` so a read view can cheaply clone the current /// snapshot while a refresh swaps in a new one. config: RwLock>, segments: RwLock>, @@ -52,8 +53,9 @@ pub struct ReadOnlyEdgeShard { /// backend-specific (see [`SegmentEnumerator`]) until an on-disk manifest exists. enumerator: Box, /// Fixed-size pool used to open segments in parallel on open/refresh and to run per-segment - /// reads in parallel. A follower has no `edge_config.json`, so it is always sized from the - /// CPU-derived default (see [`EdgeConfig::search_thread_count`]). + /// reads in parallel. Segments never carry `max_search_threads`, so it is sized from + /// `provided_config` alone: the CPU-derived default unless explicitly set (see + /// [`EdgeConfig::search_thread_count`]). search_pool: Arc, } diff --git a/lib/edge/src/read_only/refresh.rs b/lib/edge/src/read_only/refresh.rs index d40e906217..abaf1493e2 100644 --- a/lib/edge/src/read_only/refresh.rs +++ b/lib/edge/src/read_only/refresh.rs @@ -75,11 +75,26 @@ impl ReadOnlyEdgeShard { }; // 3. Re-derive the config from the current segments — a read-only follower has no - // edge_config.json, so the segments are the source of truth. No-op for an empty shard + // edge_config.json, so the segments are the source of truth. Folded over all segments + // in UUID order, so the derivation is deterministic and a segment carrying no + // information about a parameter never masks one that does. No-op for an empty shard // (the previous snapshot stays in place until segments appear). - if let Some(segment) = self.segments.read().read_handles().into_iter().next() { - let config = EdgeConfig::from_segment_config(&segment.read().segment_config); - *self.config.write() = Arc::new(config); + let derived = { + let holder = self.segments.read(); + let mut uuids = holder.uuids(); + uuids.sort_unstable(); + uuids + .into_iter() + .filter_map(|uuid| holder.segment_arc(&uuid)) + .fold(None, |acc, segment| { + Some(EdgeConfig::fold_from_segment_config( + acc, + &segment.read().segment_config, + )) + }) + }; + if let Some(derived) = derived { + *self.config.write() = Arc::new(derived); } // 4. Live-reload survivors to fold in the leader's flushed in-place appends and deletes. diff --git a/lib/edge/src/read_only/tests.rs b/lib/edge/src/read_only/tests.rs index 63abe382ec..deb75046c5 100644 --- a/lib/edge/src/read_only/tests.rs +++ b/lib/edge/src/read_only/tests.rs @@ -31,7 +31,7 @@ const VECTOR_NAME: &str = "edge-ro-test-vector"; fn test_config() -> EdgeConfig { EdgeConfig { - on_disk_payload: false, + on_disk_payload: Some(false), vectors: HashMap::from([( VECTOR_NAME.to_string(), EdgeVectorParams { @@ -45,9 +45,9 @@ fn test_config() -> EdgeConfig { }, )]), sparse_vectors: HashMap::new(), - hnsw_config: Default::default(), + hnsw_config: None, quantization_config: None, - optimizers: Default::default(), + optimizers: None, wal_options: None, max_search_threads: None, } @@ -83,6 +83,7 @@ fn open_follower(path: &std::path::Path) -> ReadOnlyEdgeShard { MmapFs, path, LocalSegmentEnumerator::new(path), + None, ) .unwrap() } @@ -308,6 +309,63 @@ fn open_without_config_derives_from_segments() { assert_eq!(expected, 10); } +/// A caller-provided config overrides tunables at open only: `vectors` still derive from the +/// segments (so reads work with a vectors-less provided config), while [`refresh`] re-derives the +/// config from the segments alone. +/// +/// [`refresh`]: ReadOnlyEdgeShard::refresh +#[test] +fn provided_config_overrides_tunables_at_open() { + let dir = tempfile::Builder::new() + .prefix("edge-ro-provided-config") + .tempdir() + .unwrap(); + + let leader = EdgeShard::new(dir.path(), test_config()).unwrap(); + upsert(&leader, 1..=10); + leader.flush(); + + // Tunables only — no vector params. Those must come from the segments. + let provided = EdgeConfig { + on_disk_payload: None, + vectors: HashMap::new(), + sparse_vectors: HashMap::new(), + hnsw_config: None, + quantization_config: None, + optimizers: None, + wal_options: None, + max_search_threads: Some(2), + }; + + let follower = ReadOnlyEdgeShard::::open_with_enumerator( + MmapFs, + dir.path(), + LocalSegmentEnumerator::new(dir.path()), + Some(provided), + ) + .unwrap(); + + let config = follower.config_snapshot(); + assert!(config.vectors.contains_key(VECTOR_NAME)); + // Unspecified tunable: falls back to the segment-derived value. + assert_eq!(config.on_disk_payload, Some(false)); + // Explicitly provided tunable: wins over the derived config. + assert_eq!(config.max_search_threads, Some(2)); + assert_eq!(exact_count(&follower), 10); + + // Refresh re-derives the config from the segments alone: the provided tunables are dropped + // (segments never carry `max_search_threads`), the segment-derived values remain. + upsert(&leader, 11..=15); + leader.flush(); + follower.refresh().unwrap(); + + let config = follower.config_snapshot(); + assert!(config.vectors.contains_key(VECTOR_NAME)); + assert_eq!(config.on_disk_payload, Some(false)); + assert_eq!(config.max_search_threads, None); + assert_eq!(exact_count(&follower), 15); +} + /// A [`SegmentEnumerator`] that scans the local `segments/` directory but hides a chosen UUID, /// standing in for a non-local enumerator (e.g. S3 / a future manifest) to exercise the injection /// seam: the follower must track exactly what the enumerator reports. @@ -353,6 +411,7 @@ fn follower_uses_injected_enumerator() { segments_path, exclude: hidden, }, + None, ) .unwrap(); assert_eq!(follower.segments_count(), all_segments.len() - 1); diff --git a/lib/edge/tests/config_merge.rs b/lib/edge/tests/config_merge.rs new file mode 100644 index 0000000000..b56dcfb9a3 --- /dev/null +++ b/lib/edge/tests/config_merge.rs @@ -0,0 +1,203 @@ +//! Reloading an existing shard with a partially specified config: unspecified (`None`) +//! parameters keep their persisted values, explicitly provided ones overwrite them. + +use std::num::NonZero; + +use edge::{Distance, EdgeConfig, EdgeOptimizersConfig, EdgeShard, EdgeVectorParams, WalOptions}; +use segment::types::HnswConfig; + +const VECTOR_NAME: &str = "edge-config-merge-test-vector"; + +fn custom_hnsw_config() -> HnswConfig { + HnswConfig { + m: 24, + ..HnswConfig::default() + } +} + +fn custom_optimizers_config() -> EdgeOptimizersConfig { + EdgeOptimizersConfig { + default_segment_number: Some(3), + ..EdgeOptimizersConfig::default() + } +} + +fn vector_params() -> EdgeVectorParams { + EdgeVectorParams { + size: 1, + distance: Distance::Dot, + quantization_config: None, + multivector_config: None, + datatype: None, + on_disk: None, + hnsw_config: None, + } +} + +fn custom_wal_options() -> WalOptions { + WalOptions { + segment_capacity: 4 * 1024 * 1024, + segment_queue_len: 0, + retain_closed: NonZero::new(1).unwrap(), + } +} + +fn full_config() -> EdgeConfig { + EdgeConfig::builder() + .vector(VECTOR_NAME, vector_params()) + .on_disk_payload(false) + .hnsw_config(custom_hnsw_config()) + .optimizers(custom_optimizers_config()) + .wal_options(custom_wal_options()) + .max_search_threads(3) + .build() +} + +/// Only vectors specified; everything else left unspecified. +fn vectors_only_config() -> EdgeConfig { + EdgeConfig::builder() + .vector(VECTOR_NAME, vector_params()) + .build() +} + +#[test] +fn reload_with_unspecified_params_keeps_persisted_values() { + let dir = tempfile::Builder::new() + .prefix("edge-config-merge-keep") + .tempdir() + .unwrap(); + + drop(EdgeShard::new(dir.path(), full_config()).unwrap()); + + let shard = EdgeShard::load(dir.path(), Some(vectors_only_config())).unwrap(); + let config = shard.config().clone(); + assert_eq!(config.on_disk_payload, Some(false)); + assert_eq!(config.hnsw_config, Some(custom_hnsw_config())); + assert_eq!(config.optimizers, Some(custom_optimizers_config())); + assert_eq!(config.wal_options, Some(custom_wal_options())); + assert_eq!(config.max_search_threads, Some(3)); + drop(shard); + + // The merged config is persisted: a plain reload sees the same values. + let shard = EdgeShard::load(dir.path(), None).unwrap(); + let config = shard.config().clone(); + assert_eq!(config.on_disk_payload, Some(false)); + assert_eq!(config.hnsw_config, Some(custom_hnsw_config())); + assert_eq!(config.optimizers, Some(custom_optimizers_config())); + assert_eq!(config.wal_options, Some(custom_wal_options())); + assert_eq!(config.max_search_threads, Some(3)); +} + +#[test] +fn reload_with_explicit_params_overwrites_persisted_values() { + let dir = tempfile::Builder::new() + .prefix("edge-config-merge-overwrite") + .tempdir() + .unwrap(); + + drop(EdgeShard::new(dir.path(), full_config()).unwrap()); + + let new_hnsw_config = HnswConfig { + m: 48, + ..HnswConfig::default() + }; + let provided = EdgeConfig::builder() + .vector(VECTOR_NAME, vector_params()) + .hnsw_config(new_hnsw_config) + .build(); + + let shard = EdgeShard::load(dir.path(), Some(provided)).unwrap(); + let config = shard.config().clone(); + // Explicitly provided: overwritten. + assert_eq!(config.hnsw_config, Some(new_hnsw_config)); + // Unspecified: kept from the persisted config. + assert_eq!(config.on_disk_payload, Some(false)); + assert_eq!(config.optimizers, Some(custom_optimizers_config())); + assert_eq!(config.wal_options, Some(custom_wal_options())); + assert_eq!(config.max_search_threads, Some(3)); +} + +/// Reloading an existing shard with a tunables-only config (no vector params at all): the vector +/// definitions are inherited from the persisted config instead of failing the compatibility +/// check, while the provided tunables still apply. +#[test] +fn reload_with_tunables_only_config_inherits_vectors() { + let dir = tempfile::Builder::new() + .prefix("edge-config-merge-no-vectors") + .tempdir() + .unwrap(); + + drop(EdgeShard::new(dir.path(), full_config()).unwrap()); + + let provided = EdgeConfig::builder().max_search_threads(2).build(); + let shard = EdgeShard::load(dir.path(), Some(provided)).unwrap(); + let config = shard.config().clone(); + assert!(config.vectors.contains_key(VECTOR_NAME)); + assert_eq!(config.max_search_threads, Some(2)); + assert_eq!(config.hnsw_config, Some(custom_hnsw_config())); +} + +/// With no persisted config either, the vector definitions come from the segments themselves — +/// the last layer of the provided → persisted → derived-from-segments → default chain. +#[test] +fn reload_without_persisted_config_derives_vectors_from_segments() { + let dir = tempfile::Builder::new() + .prefix("edge-config-merge-derived") + .tempdir() + .unwrap(); + + drop(EdgeShard::new(dir.path(), full_config()).unwrap()); + fs_err::remove_file(dir.path().join("edge_config.json")).unwrap(); + + let provided = EdgeConfig::builder().max_search_threads(2).build(); + let shard = EdgeShard::load(dir.path(), Some(provided)).unwrap(); + let config = shard.config().clone(); + assert!(config.vectors.contains_key(VECTOR_NAME)); + assert_eq!(config.max_search_threads, Some(2)); + // Derived from the segments: the shard was created with in-RAM payload storage. + assert_eq!(config.on_disk_payload, Some(false)); +} + +/// Provided vector params that don't match the stored data (here: a different vector size) must +/// fail the load instead of silently reconfiguring the shard. +#[test] +fn reload_with_incompatible_vectors_fails() { + let dir = tempfile::Builder::new() + .prefix("edge-config-merge-incompatible") + .tempdir() + .unwrap(); + + drop(EdgeShard::new(dir.path(), full_config()).unwrap()); + + let incompatible = EdgeConfig::builder() + .vector( + VECTOR_NAME, + EdgeVectorParams { + size: 2, + distance: Distance::Dot, + quantization_config: None, + multivector_config: None, + datatype: None, + on_disk: None, + hnsw_config: None, + }, + ) + .build(); + let err = EdgeShard::load(dir.path(), Some(incompatible)).unwrap_err(); + assert!(err.to_string().contains("incompatible"), "{err}"); +} + +#[test] +fn unspecified_params_resolve_to_defaults_on_new_shard() { + let dir = tempfile::Builder::new() + .prefix("edge-config-merge-defaults") + .tempdir() + .unwrap(); + + let shard = EdgeShard::new(dir.path(), vectors_only_config()).unwrap(); + let config = shard.config().clone(); + assert_eq!(config.on_disk_payload, None); + assert!(config.on_disk_payload()); + assert_eq!(config.hnsw_config(), HnswConfig::default()); + assert_eq!(config.optimizers(), EdgeOptimizersConfig::default()); +} diff --git a/lib/edge/tests/wal_options.rs b/lib/edge/tests/wal_options.rs index 1544ac0541..e75e42c801 100644 --- a/lib/edge/tests/wal_options.rs +++ b/lib/edge/tests/wal_options.rs @@ -147,8 +147,19 @@ fn reload_with_larger_wal_capacity_after_upsert() { assert_eq!(shard.info().points_count, 1); } - // Phase 2: reload with default 32 MiB WAL (overwrites persisted config). - let shard = EdgeShard::load(dir.path(), Some(default_config())).unwrap(); + // Phase 2: reload with explicit default 32 MiB WAL options, overwriting the + // persisted small ones. (Leaving wal_options unspecified would keep them.) + let config = base_builder().wal_options(WalOptions::default()).build(); + let shard = EdgeShard::load(dir.path(), Some(config)).unwrap(); + assert_eq!( + shard + .config() + .wal_options + .as_ref() + .unwrap() + .segment_capacity, + WalOptions::default().segment_capacity, + ); assert_eq!( shard.info().points_count, 1, diff --git a/lib/edge/tools/shard_query/src/main.rs b/lib/edge/tools/shard_query/src/main.rs index bff99c58b5..632c42fb12 100644 --- a/lib/edge/tools/shard_query/src/main.rs +++ b/lib/edge/tools/shard_query/src/main.rs @@ -529,7 +529,7 @@ where // No edge_config.json: `ReadOnlyEdgeShard` derives its config from the segments and discovers // them via the manifest. `prefix` is passed only as the shard's (logical) path label. - let shard = ReadOnlyEdgeShard::>>::open(cached_fs, prefix) + let shard = ReadOnlyEdgeShard::>>::open(cached_fs, prefix, None) .context("failed to open read-only edge shard over object storage")?; log::info!("opened shard with {} segment(s)", shard.segments_count());