upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix: Non-blocking payload index building (#4941)

Browse Source 
* separate index creation into build and apply

* check version in Segment impl of `build_field_index`

* add wait to issues test

* fix consensus tests
This commit is contained in:
Luis Cossío
2024-08-23 08:30:45 -04:00
committed by generall
parent 9e8ba48bf1
commit 96158c6f27
10 changed files with 192 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
lib/collection/src/collection_manager/holders/proxy_segment.rs
View File

@@ -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<Option<(PayloadFieldSchema, Vec<FieldIndex>)>> {
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<FieldIndex>,
) -> OperationResult<bool> {
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)
}

6
lib/collection/src/collection_manager/holders/segment_holder.rs
View File

@@ -410,13 +410,15 @@ impl<'s> SegmentHolder {
pub fn apply_segments<F>(&self, mut f: F) -> OperationResult<usize>
where
F: FnMut(&mut RwLockWriteGuard<dyn SegmentEntry + 'static>) -> OperationResult<bool>,
F: FnMut(
&mut RwLockUpgradableReadGuard<dyn SegmentEntry + 'static>,
) -> OperationResult<bool>,
{
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)

14
lib/collection/src/collection_manager/segments_updater.rs
View File

@@ -335,7 +335,15 @@ pub(crate) fn create_field_index(
) -> CollectionResult<usize> {
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<usize> {
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)
}

27
lib/segment/src/entry/entry_point.rs
View File

@@ -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<bool>;
/// 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<Option<(PayloadFieldSchema, Vec<FieldIndex>)>>;
/// 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<FieldIndex>,
) -> OperationResult<bool>;
/// 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<bool>;
) -> OperationResult<bool> {
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<PayloadKeyType, PayloadFieldSchema>;

28
lib/segment/src/index/payload_index_base.rs
View File

@@ -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<PayloadKeyType, PayloadFieldSchema>;
/// Build the index, if not built before, taking the caller by reference only
fn build_index(
&self,
field: PayloadKeyTypeRef,
payload_schema: &PayloadFieldSchema,
) -> OperationResult<Option<Vec<FieldIndex>>>;
/// Apply already built indexes
fn apply_index(
&mut self,
field: PayloadKeyType,
payload_schema: PayloadFieldSchema,
field_index: Vec<FieldIndex>,
) -> OperationResult<()>;
/// Mark field as one which should be indexed
fn set_indexed(
&mut self,
field: PayloadKeyTypeRef,
payload_schema: impl Into<PayloadFieldSchema>,
) -> 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<()>;

22
lib/segment/src/index/plain_payload_index.rs
View File

@@ -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<PayloadFieldSchema>,
) -> OperationResult<()> {
let payload_schema = payload_schema.into();
fn build_index(
&self,
_field: PayloadKeyTypeRef,
_payload_schema: &PayloadFieldSchema,
) -> OperationResult<Option<Vec<FieldIndex>>> {
Ok(Some(Vec::new()))
}
fn apply_index(
&mut self,
field: PayloadKeyType,
payload_schema: PayloadFieldSchema,
_field_index: Vec<FieldIndex>,
) -> 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 {

59
lib/segment/src/index/struct_payload_index.rs
View File

@@ -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<Vec<FieldIndex>> {
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<Vec<FieldIndex>> {
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<PayloadFieldSchema>,
) -> 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<Option<Vec<FieldIndex>>> {
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<FieldIndex>,
) -> OperationResult<()> {
self.field_indexes.insert(field.clone(), field_index);
self.config.indexed_fields.insert(field, payload_schema);
self.save_config()?;
Ok(())

53
lib/segment/src/segment/entry.rs
View File

@@ -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<bool> {
self.handle_segment_version_and_failure(op_num, |segment| match field_type {
) -> OperationResult<Option<(PayloadFieldSchema, Vec<FieldIndex>)>> {
// 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<FieldIndex>,
) -> OperationResult<bool> {
self.handle_segment_version_and_failure(op_num, |segment| {
segment
.payload_index
.borrow_mut()
.apply_index(key, schema, field_index)?;
Ok(true)
})
}

1
tests/consensus_tests/fixtures.py
View File

@@ -126,6 +126,7 @@ def create_field_index(
"field_schema": field_schema,
},
headers=headers,
params={"wait": "true"}
)
assert_http_ok(r_batch)

9
tests/consensus_tests/test_issues_api.py
View File

@@ -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