From 2bb287d693c8f5e10158844f4294fcb565750c7d Mon Sep 17 00:00:00 2001 From: Andrey Vasnetsov Date: Tue, 7 Jul 2026 11:45:02 +0200 Subject: [PATCH] Make EdgeConfig tunables optional with a layered fallback chain (#9714) Tunable EdgeConfig parameters (on_disk_payload, hnsw_config, optimizers) are now Option, and every tunable resolves through the fallback chain provided -> persisted -> derived from segments -> default when loading an existing shard. Leaving a parameter unspecified keeps the shard as it is; an explicit value overwrites it and existing segments converge to it through the optimizers. vectors/sparse_vectors are excluded from overwrite semantics: an empty map inherits the persisted/segment-derived definitions, a non-empty map is validated for compatibility against the loaded segments (size, distance, multivector, datatype, sparse modifier) and fails the load on mismatch. The derived layer folds over all segments in UUID order instead of taking an arbitrary first segment, so a plain appendable segment (which carries no HNSW parameters) can never mask an indexed segment's actual build parameters. Previously a lost edge_config.json could resolve unspecified HNSW params to compiled-in defaults and silently trigger a full re-index via ConfigMismatchOptimizer. The read-only follower accepts an optional config on open: provided tunables are applied once over the segment-derived config (vectors always come from the segments), and refresh re-derives from segments alone. Co-authored-by: Claude Fable 5 --- lib/edge/python/qdrant_edge.pyi | 21 ++- lib/edge/python/src/config/mod.rs | 18 +-- lib/edge/src/builders/edge_config.rs | 20 +-- lib/edge/src/config/shard.rs | 205 +++++++++++++++++++------ lib/edge/src/lib.rs | 108 ++++++++----- lib/edge/src/optimize.rs | 15 +- lib/edge/src/read_only/lifecycle.rs | 61 ++++++-- lib/edge/src/read_only/mod.rs | 10 +- lib/edge/src/read_only/refresh.rs | 23 ++- lib/edge/src/read_only/tests.rs | 65 +++++++- lib/edge/tests/config_merge.rs | 203 ++++++++++++++++++++++++ lib/edge/tests/wal_options.rs | 15 +- lib/edge/tools/shard_query/src/main.rs | 2 +- 13 files changed, 615 insertions(+), 151 deletions(-) create mode 100644 lib/edge/tests/config_merge.rs 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());