diff --git a/lib/collection/src/collection_manager/holders/proxy_segment.rs b/lib/collection/src/collection_manager/holders/proxy_segment.rs index 943187bba0..3656435cbf 100644 --- a/lib/collection/src/collection_manager/holders/proxy_segment.rs +++ b/lib/collection/src/collection_manager/holders/proxy_segment.rs @@ -14,7 +14,7 @@ use segment::data_types::order_by::OrderValue; use segment::data_types::query_context::{QueryContext, SegmentQueryContext}; use segment::data_types::vectors::{QueryVector, Vector}; use segment::entry::entry_point::SegmentEntry; -use segment::index::field_index::CardinalityEstimation; +use segment::index::field_index::{CardinalityEstimation, FieldIndex}; use segment::json_path::JsonPath; use segment::telemetry::SegmentTelemetry; use segment::types::{ @@ -956,30 +956,47 @@ impl SegmentEntry for ProxySegment { .delete_field_index(op_num, key) } - fn create_field_index( - &mut self, - op_num: u64, + fn build_field_index( + &self, + op_num: SeqNumberType, key: PayloadKeyTypeRef, - field_schema: Option<&PayloadFieldSchema>, + field_type: Option<&PayloadFieldSchema>, + ) -> OperationResult)>> { + if self.version() > op_num { + return Ok(None); + } + + self.write_segment + .get() + .read() + .build_field_index(op_num, key, field_type) + } + + fn apply_field_index( + &mut self, + op_num: SeqNumberType, + key: PayloadKeyType, + field_schema: PayloadFieldSchema, + field_index: Vec, ) -> OperationResult { if self.version() > op_num { return Ok(false); } - self.write_segment - .get() - .write() - .create_field_index(op_num, key, field_schema)?; - let indexed_fields = self.write_segment.get().read().get_indexed_fields(); - - let Some(payload_schema) = indexed_fields.get(key) else { + if !self.write_segment.get().write().apply_field_index( + op_num, + key.clone(), + field_schema.clone(), + field_index, + )? { return Ok(false); }; + // Index was updated: mark as added and deleted self.created_indexes .write() - .insert(key.to_owned(), payload_schema.to_owned()); - self.deleted_indexes.write().remove(key); + .insert(key.clone(), field_schema); + self.deleted_indexes.write().remove(&key); Ok(true) } diff --git a/lib/collection/src/collection_manager/holders/segment_holder.rs b/lib/collection/src/collection_manager/holders/segment_holder.rs index 5a15ac9c04..cf6c99c5a7 100644 --- a/lib/collection/src/collection_manager/holders/segment_holder.rs +++ b/lib/collection/src/collection_manager/holders/segment_holder.rs @@ -410,13 +410,15 @@ impl<'s> SegmentHolder { pub fn apply_segments(&self, mut f: F) -> OperationResult where - F: FnMut(&mut RwLockWriteGuard) -> OperationResult, + F: FnMut( + &mut RwLockUpgradableReadGuard, + ) -> OperationResult, { let _update_guard = self.update_tracker.update(); let mut processed_segments = 0; for (_id, segment) in self.iter() { - let is_applied = f(&mut segment.get().write())?; + let is_applied = f(&mut segment.get().upgradable_read())?; processed_segments += usize::from(is_applied); } Ok(processed_segments) diff --git a/lib/collection/src/collection_manager/segments_updater.rs b/lib/collection/src/collection_manager/segments_updater.rs index f76e17e9b4..dfa8f7fa97 100644 --- a/lib/collection/src/collection_manager/segments_updater.rs +++ b/lib/collection/src/collection_manager/segments_updater.rs @@ -335,7 +335,15 @@ pub(crate) fn create_field_index( ) -> CollectionResult { segments .apply_segments(|write_segment| { - write_segment.create_field_index(op_num, field_name, field_schema) + let Some((schema, index)) = + write_segment.build_field_index(op_num, field_name, field_schema)? + else { + return Ok(false); + }; + + write_segment.with_upgraded(|segment| { + segment.apply_field_index(op_num, field_name.to_owned(), schema, index) + }) }) .map_err(Into::into) } @@ -346,7 +354,9 @@ pub(crate) fn delete_field_index( field_name: PayloadKeyTypeRef, ) -> CollectionResult { segments - .apply_segments(|write_segment| write_segment.delete_field_index(op_num, field_name)) + .apply_segments(|write_segment| { + write_segment.with_upgraded(|segment| segment.delete_field_index(op_num, field_name)) + }) .map_err(Into::into) } diff --git a/lib/segment/src/entry/entry_point.rs b/lib/segment/src/entry/entry_point.rs index 18512bfc94..7e310271f0 100644 --- a/lib/segment/src/entry/entry_point.rs +++ b/lib/segment/src/entry/entry_point.rs @@ -10,7 +10,7 @@ use crate::data_types::named_vectors::NamedVectors; use crate::data_types::order_by::{OrderBy, OrderValue}; use crate::data_types::query_context::{QueryContext, SegmentQueryContext}; use crate::data_types::vectors::{QueryVector, Vector}; -use crate::index::field_index::CardinalityEstimation; +use crate::index::field_index::{CardinalityEstimation, FieldIndex}; use crate::json_path::JsonPath; use crate::telemetry::SegmentTelemetry; use crate::types::{ @@ -220,13 +220,36 @@ pub trait SegmentEntry { key: PayloadKeyTypeRef, ) -> OperationResult; + /// Build the field index for the key and schema, if not built before. + fn build_field_index( + &self, + op_num: SeqNumberType, + key: PayloadKeyTypeRef, + field_type: Option<&PayloadFieldSchema>, + ) -> OperationResult)>>; + + /// Apply a built index. Returns whether it was actually applied or not. + fn apply_field_index( + &mut self, + op_num: SeqNumberType, + key: PayloadKeyType, + field_schema: PayloadFieldSchema, + field_index: Vec, + ) -> OperationResult; + /// Create index for a payload field, if not exists fn create_field_index( &mut self, op_num: SeqNumberType, key: PayloadKeyTypeRef, field_schema: Option<&PayloadFieldSchema>, - ) -> OperationResult; + ) -> OperationResult { + let Some((schema, index)) = self.build_field_index(op_num, key, field_schema)? else { + return Ok(false); + }; + + self.apply_field_index(op_num, key.to_owned(), schema, index) + } /// Get indexed fields fn get_indexed_fields(&self) -> HashMap; diff --git a/lib/segment/src/index/payload_index_base.rs b/lib/segment/src/index/payload_index_base.rs index 73b34766cf..fb6fb77635 100644 --- a/lib/segment/src/index/payload_index_base.rs +++ b/lib/segment/src/index/payload_index_base.rs @@ -4,6 +4,7 @@ use std::path::{Path, PathBuf}; use common::types::PointOffsetType; use serde_json::Value; +use super::field_index::FieldIndex; use crate::common::operation_error::OperationResult; use crate::common::Flusher; use crate::index::field_index::{CardinalityEstimation, PayloadBlockCondition}; @@ -17,12 +18,37 @@ pub trait PayloadIndex { /// Get indexed fields fn indexed_fields(&self) -> HashMap; + /// Build the index, if not built before, taking the caller by reference only + fn build_index( + &self, + field: PayloadKeyTypeRef, + payload_schema: &PayloadFieldSchema, + ) -> OperationResult>>; + + /// Apply already built indexes + fn apply_index( + &mut self, + field: PayloadKeyType, + payload_schema: PayloadFieldSchema, + field_index: Vec, + ) -> OperationResult<()>; + /// Mark field as one which should be indexed fn set_indexed( &mut self, field: PayloadKeyTypeRef, payload_schema: impl Into, - ) -> OperationResult<()>; + ) -> OperationResult<()> { + let payload_schema = payload_schema.into(); + + let Some(field_index) = self.build_index(field, &payload_schema)? else { + return Ok(()); + }; + + self.apply_index(field.to_owned(), payload_schema, field_index)?; + + Ok(()) + } /// Remove index fn drop_index(&mut self, field: PayloadKeyTypeRef) -> OperationResult<()>; diff --git a/lib/segment/src/index/plain_payload_index.rs b/lib/segment/src/index/plain_payload_index.rs index c1f6219e02..3f242210c9 100644 --- a/lib/segment/src/index/plain_payload_index.rs +++ b/lib/segment/src/index/plain_payload_index.rs @@ -8,6 +8,7 @@ use common::types::{PointOffsetType, ScoredPointOffset, TelemetryDetail}; use parking_lot::Mutex; use schemars::_serde_json::Value; +use super::field_index::FieldIndex; use crate::common::operation_error::OperationResult; use crate::common::operation_time_statistics::{ OperationDurationStatistics, OperationDurationsAggregator, ScopeDurationMeasurer, @@ -83,17 +84,24 @@ impl PayloadIndex for PlainPayloadIndex { self.config.indexed_fields.clone() } - fn set_indexed( - &mut self, - field: PayloadKeyTypeRef, - payload_schema: impl Into, - ) -> OperationResult<()> { - let payload_schema = payload_schema.into(); + fn build_index( + &self, + _field: PayloadKeyTypeRef, + _payload_schema: &PayloadFieldSchema, + ) -> OperationResult>> { + Ok(Some(Vec::new())) + } + fn apply_index( + &mut self, + field: PayloadKeyType, + payload_schema: PayloadFieldSchema, + _field_index: Vec, + ) -> OperationResult<()> { if let Some(prev_schema) = self .config .indexed_fields - .insert(field.to_owned(), payload_schema.clone()) + .insert(field, payload_schema.clone()) { // the field is already present with the same schema, no need to save the config if prev_schema == payload_schema { diff --git a/lib/segment/src/index/struct_payload_index.rs b/lib/segment/src/index/struct_payload_index.rs index b1cbc44759..6fe1fee777 100644 --- a/lib/segment/src/index/struct_payload_index.rs +++ b/lib/segment/src/index/struct_payload_index.rs @@ -106,7 +106,7 @@ impl StructPayloadIndex { let mut field_indexes: IndexesMap = Default::default(); for (field, payload_schema) in &self.config.indexed_fields { - let field_index = self.load_from_db(field, payload_schema.to_owned())?; + let field_index = self.load_from_db(field, payload_schema)?; field_indexes.insert(field.clone(), field_index); } self.field_indexes = field_indexes; @@ -116,11 +116,11 @@ impl StructPayloadIndex { fn load_from_db( &self, field: PayloadKeyTypeRef, - payload_schema: PayloadFieldSchema, + payload_schema: &PayloadFieldSchema, ) -> OperationResult> { let mut indexes = self - .selector(&payload_schema) - .new_index(field, &payload_schema)?; + .selector(payload_schema) + .new_index(field, payload_schema)?; let mut is_loaded = true; for ref mut index in indexes.iter_mut() { @@ -179,12 +179,12 @@ impl StructPayloadIndex { pub fn build_field_indexes( &self, field: PayloadKeyTypeRef, - payload_schema: PayloadFieldSchema, + payload_schema: &PayloadFieldSchema, ) -> OperationResult> { let payload_storage = self.payload.borrow(); let mut builders = self - .selector(&payload_schema) - .index_builder(field, &payload_schema)?; + .selector(payload_schema) + .index_builder(field, payload_schema)?; for index in &mut builders { index.init()?; @@ -204,16 +204,6 @@ impl StructPayloadIndex { .collect() } - fn build_and_save( - &mut self, - field: PayloadKeyTypeRef, - payload_schema: PayloadFieldSchema, - ) -> OperationResult<()> { - let field_indexes = self.build_field_indexes(field, payload_schema)?; - self.field_indexes.insert(field.clone(), field_indexes); - Ok(()) - } - /// Number of available points /// /// - excludes soft deleted points @@ -429,25 +419,34 @@ impl PayloadIndex for StructPayloadIndex { self.config.indexed_fields.clone() } - fn set_indexed( - &mut self, + fn build_index( + &self, field: PayloadKeyTypeRef, - payload_schema: impl Into, - ) -> OperationResult<()> { - let payload_schema = payload_schema.into(); - - if let Some(prev_schema) = self - .config - .indexed_fields - .insert(field.to_owned(), payload_schema.clone()) - { + payload_schema: &PayloadFieldSchema, + ) -> OperationResult>> { + if let Some(prev_schema) = self.config.indexed_fields.get(field) { // the field is already indexed with the same schema // no need to rebuild index and to save the config if prev_schema == payload_schema { - return Ok(()); + return Ok(None); } } - self.build_and_save(field, payload_schema)?; + + let indexes = self.build_field_indexes(field, payload_schema)?; + + Ok(Some(indexes)) + } + + fn apply_index( + &mut self, + field: PayloadKeyType, + payload_schema: PayloadFieldSchema, + field_index: Vec, + ) -> OperationResult<()> { + self.field_indexes.insert(field.clone(), field_index); + + self.config.indexed_fields.insert(field, payload_schema); + self.save_config()?; Ok(()) diff --git a/lib/segment/src/segment/entry.rs b/lib/segment/src/segment/entry.rs index 0b38a7eb65..900c23fa1c 100644 --- a/lib/segment/src/segment/entry.rs +++ b/lib/segment/src/segment/entry.rs @@ -19,7 +19,7 @@ use crate::data_types::order_by::{OrderBy, OrderValue}; use crate::data_types::query_context::{QueryContext, SegmentQueryContext}; use crate::data_types::vectors::{QueryVector, Vector}; use crate::entry::entry_point::SegmentEntry; -use crate::index::field_index::CardinalityEstimation; +use crate::index::field_index::{CardinalityEstimation, FieldIndex}; use crate::index::{PayloadIndex, VectorIndex}; use crate::json_path::JsonPath; use crate::segment::{ @@ -639,32 +639,59 @@ impl SegmentEntry for Segment { }) } - fn create_field_index( - &mut self, - op_num: u64, + fn build_field_index( + &self, + op_num: SeqNumberType, key: PayloadKeyTypeRef, field_type: Option<&PayloadFieldSchema>, - ) -> OperationResult { - self.handle_segment_version_and_failure(op_num, |segment| match field_type { + ) -> OperationResult)>> { + // Check version without updating it + if self.version.unwrap_or(0) > op_num { + return Ok(None); + } + + match field_type { Some(schema) => { - segment + let res = self .payload_index .borrow_mut() - .set_indexed(key, schema.clone())?; - Ok(true) + .build_index(key, schema)? + .map(|field_index| (schema.to_owned(), field_index)); + + Ok(res) } - None => match segment.infer_from_payload_data(key)? { + None => match self.infer_from_payload_data(key)? { None => Err(TypeInferenceError { field_name: key.clone(), }), Some(schema_type) => { - segment + let schema = schema_type.into(); + + let res = self .payload_index .borrow_mut() - .set_indexed(key, schema_type)?; - Ok(true) + .build_index(key, &schema)? + .map(|field_index| (schema, field_index)); + + Ok(res) } }, + } + } + + fn apply_field_index( + &mut self, + op_num: SeqNumberType, + key: PayloadKeyType, + schema: PayloadFieldSchema, + field_index: Vec, + ) -> OperationResult { + self.handle_segment_version_and_failure(op_num, |segment| { + segment + .payload_index + .borrow_mut() + .apply_index(key, schema, field_index)?; + Ok(true) }) } diff --git a/tests/consensus_tests/fixtures.py b/tests/consensus_tests/fixtures.py index f3d316f542..0fbdd508ec 100644 --- a/tests/consensus_tests/fixtures.py +++ b/tests/consensus_tests/fixtures.py @@ -126,6 +126,7 @@ def create_field_index( "field_schema": field_schema, }, headers=headers, + params={"wait": "true"} ) assert_http_ok(r_batch) diff --git a/tests/consensus_tests/test_issues_api.py b/tests/consensus_tests/test_issues_api.py index e229db2b0d..752454f4a9 100644 --- a/tests/consensus_tests/test_issues_api.py +++ b/tests/consensus_tests/test_issues_api.py @@ -95,6 +95,8 @@ def test_unindexed_field_is_gone_when_indexing(setup_with_big_collection): # check the issue is now active issue_present = False + solution = None + timestamp = None for issue in issues: if issue["id"] == expected_issue_code: issue_present = True @@ -102,7 +104,9 @@ def test_unindexed_field_is_gone_when_indexing(setup_with_big_collection): timestamp = issue["timestamp"] break assert issue_present - + assert solution is not None + assert timestamp is not None + # search again search_with_city_filter(uri) @@ -116,6 +120,8 @@ def test_unindexed_field_is_gone_when_indexing(setup_with_big_collection): assert timestamp == issue["timestamp"] break assert issue_present + assert solution is not None + assert timestamp is not None assert solution == { "method": "PUT", @@ -128,6 +134,7 @@ def test_unindexed_field_is_gone_when_indexing(setup_with_big_collection): response = requests.request( method=solution["method"], url=uri + solution["uri"], + params={"wait": "true"}, json=solution["body"], ) assert response.ok