diff --git a/crates/persistence/src/backends/mongodb/missing_fingerprint.rs b/crates/persistence/src/backends/mongodb/missing_fingerprint.rs index 205134a86e..4ac9fdb864 100644 --- a/crates/persistence/src/backends/mongodb/missing_fingerprint.rs +++ b/crates/persistence/src/backends/mongodb/missing_fingerprint.rs @@ -49,7 +49,7 @@ pub(super) enum MissingFreshness { } impl MissingFreshness { - fn allows(&self, code: &str) -> bool { + pub(super) fn allows(&self, code: &str) -> bool { match self { Self::All => true, Self::Only(codes) => codes.contains(code), @@ -167,27 +167,23 @@ impl MongoBackend { stamp } - /// Whether `:missing=true` on `code` may read `missing_params`: only - /// after this registry's probe found the parameter fresh. A missing or - /// outdated probe result (a search path that did not probe, or a - /// definition change since) is not trusted. - pub(super) fn missing_param_fresh( + /// Which codes of `resource_type` may read `missing_params`: only those + /// this registry's probe found fresh. A missing or outdated probe result + /// (a search path that did not probe, or a definition change since) is + /// `None`, which trusts no code. + pub(super) fn missing_freshness( &self, tenant_id: &str, resource_type: &str, - code: &str, - ) -> bool { + ) -> Option { let key = (tenant_id.to_string(), resource_type.to_string()); - let Some((fingerprint, freshness)) = self.missing_meta_fresh.read().get(&key).cloned() - else { - return false; - }; - freshness.allows(code) - && fingerprint - == self - .tenant_extractor(tenant_id) - .missing_fingerprint(resource_type) - .fingerprint + let (fingerprint, freshness) = self.missing_meta_fresh.read().get(&key).cloned()?; + (fingerprint + == self + .tenant_extractor(tenant_id) + .missing_fingerprint(resource_type) + .fingerprint) + .then_some(freshness) } /// Probes `idx_resources_missing_params_fp` before a `:missing=true` diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index dcba6cbe1c..2973d2e72f 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -1043,6 +1043,7 @@ impl MongoBackend { let db = self.get_database().await?; let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); let tenant_id = tenant.tenant_id().as_str(); + let missing_support = self.classify_missing(tenant_id, &query.resource_type, query); let cursor = if let Some(cursor_str) = &query.cursor { Some(PageCursor::decode(cursor_str).map_err(|_| { @@ -1084,7 +1085,14 @@ impl MongoBackend { } self.build_sort_document(query, previous_mode)?; let total = self - .count_standard_with_facts(tenant, query, facts.count_selection(), &mut probe_facts) + .count_standard_with_facts( + tenant, + &db, + query, + &missing_support, + facts.count_selection(), + &mut probe_facts, + ) .await?; probe_facts.record(); return Ok( @@ -1101,6 +1109,7 @@ impl MongoBackend { &db, tenant_id, query, + &missing_support, execution.filter_purpose, execution.page_policy, index_cursor_filter.as_ref(), @@ -1108,7 +1117,7 @@ impl MongoBackend { ) .await?; let (filter_plan, matched_ids) = self - .plan_or_matching_ids(&db, tenant_id, query, filter_plan) + .plan_or_matching_ids(&db, tenant_id, query, &missing_support, filter_plan) .await?; record_filter_strategy( "filter_strategy", @@ -1159,6 +1168,7 @@ impl MongoBackend { .search_param_sorted( tenant, query, + &missing_support, &db, tenant_id, matched_ids, @@ -1175,6 +1185,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + &missing_support, matched_ids.as_ref(), None, )?; @@ -1199,6 +1210,7 @@ impl MongoBackend { &db, tenant_id, query, + &missing_support, plan, scope_filter, cursor_condition, @@ -1212,7 +1224,14 @@ impl MongoBackend { internal_error("Batched page requires an ID sort direction".to_owned()) })?; let match_filter = if combine_total { - self.build_resource_filter(tenant_id, &query.resource_type, query, None, None)? + self.build_resource_filter( + tenant_id, + &query.resource_type, + query, + &missing_support, + None, + None, + )? } else { filter.clone() }; @@ -1340,7 +1359,9 @@ impl MongoBackend { Some( self.count_standard_with_facts( tenant, + &db, query, + &missing_support, facts.count_selection(), &mut probe_facts, ) @@ -1640,50 +1661,115 @@ pub(super) fn records_missing_metadata(name: &str) -> bool { !matches!(name, "_id" | "_lastUpdated") } +/// Which `:missing` parameters of one query `missing_params` answers. +/// +/// Classified once per query by [`MongoBackend::classify_missing`] and passed +/// through planning and execution, so every stage reads the same answer: one +/// registry read and one extractor per query instead of one per check, and a +/// registry reload mid-query cannot make the planner and the executor +/// disagree (#2007). The set holds the `(name, type)` pairs that pass both +/// the definition check and the pre-search freshness probe for the resource +/// type (#1999). +#[derive(Debug, Clone, Default)] +pub(super) struct MissingSupport { + indexed: HashSet<(String, SearchParamType)>, +} + +impl MissingSupport { + /// For a query with no `:missing` parameter, or one whose parameters are + /// all `_id` / `_lastUpdated`: nothing is answered from `missing_params`. + pub(super) fn none() -> Self { + Self::default() + } + + /// Whether `param`'s `:missing` is answered from `missing_params`, for + /// either value. + pub(super) fn supports(&self, param: &SearchParameter) -> bool { + self.indexed + .contains(&(param.name.clone(), param.param_type)) + } + + /// Whether `param` is a `:missing=true` that `missing_params` answers on + /// the resource document. + pub(super) fn is_resource_parameter(&self, param: &SearchParameter) -> bool { + records_missing_metadata(¶m.name) + && matches!(param.modifier, Some(SearchModifier::Missing)) + && param + .values + .first() + .is_some_and(|value| value.value == "true") + && self.supports(param) + } +} + +/// True when the query carries a predicate that lives on the resource +/// document rather than in the search index. +fn has_resource_level_params(query: &SearchQuery, missing_support: &MissingSupport) -> bool { + query.parameters.iter().any(|p| { + matches!(p.name.as_str(), "_id" | "_lastUpdated") + || missing_support.is_resource_parameter(p) + }) +} + impl MongoBackend { - fn supports_indexed_missing( + /// Classifies `query`'s `:missing` parameters against `resource_type` once. + pub(super) fn classify_missing( &self, tenant_id: &str, resource_type: &str, - param: &SearchParameter, - ) -> bool { + query: &SearchQuery, + ) -> MissingSupport { // Until `idx_resources_missing_params` is built (#2001) every // parameter takes the value-index route, as an unsupported one does: // same results, no full `resources` scan, and no hint naming an // index that does not exist yet. if !self.missing_params_index_ready() { - return false; + return MissingSupport::none(); + } + let params: Vec<&SearchParameter> = query + .parameters + .iter() + .filter(|param| matches!(param.modifier, Some(SearchModifier::Missing))) + .collect(); + if params.is_empty() { + return MissingSupport::none(); } - let registry = self.tenant_registry(tenant_id); - let definition = { + let definitions: Vec<_> = { + let registry = self.tenant_registry(tenant_id); let registry = registry.read(); - registry.get_param(resource_type, ¶m.name).or_else(|| { - helios_fhir::search::ABSTRACT_BASE_TYPES - .iter() - .find_map(|base| registry.get_param(base, ¶m.name)) - }) + params + .into_iter() + .filter_map(|param| { + registry + .get_param(resource_type, ¶m.name) + .or_else(|| { + helios_fhir::search::ABSTRACT_BASE_TYPES + .iter() + .find_map(|base| registry.get_param(base, ¶m.name)) + }) + .filter(|definition| definition.param_type == param.param_type) + .map(|definition| (param, definition)) + }) + .collect() }; - definition.is_some_and(|definition| { - definition.param_type == param.param_type - && self - .tenant_extractor(tenant_id) - .supports_missing(resource_type, &definition) - }) && self.missing_param_fresh(tenant_id, resource_type, ¶m.name) - } - - fn is_missing_resource_parameter( - &self, - tenant_id: &str, - resource_type: &str, - param: &SearchParameter, - ) -> bool { - records_missing_metadata(¶m.name) - && matches!(param.modifier, Some(SearchModifier::Missing)) - && param - .values - .first() - .is_some_and(|value| value.value == "true") - && self.supports_indexed_missing(tenant_id, resource_type, param) + if definitions.is_empty() { + return MissingSupport::none(); + } + // Only parameters the pre-search probe found fresh (#1999). + let Some(freshness) = self.missing_freshness(tenant_id, resource_type) else { + return MissingSupport::none(); + }; + let extractor = self.tenant_extractor(tenant_id); + MissingSupport { + indexed: definitions + .into_iter() + .filter(|(param, definition)| { + freshness.allows(¶m.name) + && extractor.supports_missing(resource_type, definition) + }) + .map(|(param, _)| (param.name.clone(), param.param_type)) + .collect(), + } } /// Chooses between `plan` and the candidate id set. `IndexIntersection` @@ -1696,6 +1782,7 @@ impl MongoBackend { db: &mongodb::Database, tenant_id: &str, query: &SearchQuery, + missing_support: &MissingSupport, plan: Option, ) -> StorageResult<(Option, Option>)> { match plan { @@ -1704,14 +1791,15 @@ impl MongoBackend { // ID set; ordinary type scans retain the upstream selection below. Some(plan) if !plan.scans_type() - || query.parameters.iter().any(|param| { - self.is_missing_resource_parameter(tenant_id, &query.resource_type, param) - }) => + || query + .parameters + .iter() + .any(|param| missing_support.is_resource_parameter(param)) => { Ok((Some(plan), None)) } plan => match self - .matching_resource_ids(db, tenant_id, &query.resource_type, query) + .matching_resource_ids(db, tenant_id, &query.resource_type, query, missing_support) .await { // Complement sets are not capped while they are collected; @@ -1738,6 +1826,7 @@ impl MongoBackend { db: &mongodb::Database, tenant_id: &str, query: &SearchQuery, + missing_support: &MissingSupport, plan: search_plan::ResourceBatchPlan, scope_filter: Document, cursor_condition: Option, @@ -1768,7 +1857,10 @@ impl MongoBackend { if let Some(ids) = run.try_finish_sparse_page(db).await? { ids } else { - match self.sparse_page_id_set(db, tenant_id, query).await? { + match self + .sparse_page_id_set(db, tenant_id, query, missing_support) + .await? + { Some(id_set) => { run.record_fallback(live_matches, id_set.len()); let filter = with_cursor_condition( @@ -1776,6 +1868,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + missing_support, Some(&id_set), None, )?, @@ -1820,9 +1913,10 @@ impl MongoBackend { db: &mongodb::Database, tenant_id: &str, query: &SearchQuery, + missing_support: &MissingSupport, ) -> StorageResult>> { match self - .matching_resource_ids(db, tenant_id, &query.resource_type, query) + .matching_resource_ids(db, tenant_id, &query.resource_type, query, missing_support) .await { Ok(Some(ids)) if search_plan::id_set_fits_a_page_filter(&ids) => Ok(Some(ids)), @@ -1874,8 +1968,11 @@ impl MongoBackend { ) -> StorageResult { let selection = self.execution_facts(query).count_selection(); let mut facts = ProbeFacts::default(); + let db = self.get_database().await?; + let missing_support = + self.classify_missing(tenant.tenant_id().as_str(), &query.resource_type, query); let result = self - .count_standard_with_facts(tenant, query, selection, &mut facts) + .count_standard_with_facts(tenant, &db, query, &missing_support, selection, &mut facts) .await; facts.record(); result @@ -1885,19 +1982,21 @@ impl MongoBackend { async fn count_standard_with_facts( &self, tenant: &TenantContext, + db: &mongodb::Database, query: &SearchQuery, + missing_support: &MissingSupport, selection: BatchSelection, facts: &mut ProbeFacts, ) -> StorageResult { - let db = self.get_database().await?; let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); let tenant_id = tenant.tenant_id().as_str(); let filter_plan = self .resource_filter_plan( - &db, + db, tenant_id, query, + missing_support, SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -1905,7 +2004,7 @@ impl MongoBackend { ) .await?; let (filter_plan, matched_ids) = self - .plan_or_matching_ids(&db, tenant_id, query, filter_plan) + .plan_or_matching_ids(db, tenant_id, query, missing_support, filter_plan) .await?; record_filter_strategy( "total_filter_strategy", @@ -1917,6 +2016,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + missing_support, matched_ids.as_ref(), None, )?; @@ -1928,7 +2028,7 @@ impl MongoBackend { BatchSelection::EstimatedCount => "index_estimate", _ => "exact_batches", }); - return Ok(plan.execute(&db, filter, None, selection).await?.count); + return Ok(plan.execute(db, filter, None, selection).await?.count); } if let Some(plan) = filter_plan { record_total_strategy("exact_aggregation"); @@ -2091,13 +2191,22 @@ impl MongoBackend { // just the current top-level page, so the contained pipeline // can discard overlaps before its offset and count stages. let top_filter = if query.contained_return == ContainedReturn::Container { + let missing_support = + self.classify_missing(tenant_id, contained_type, &top_query); let matched_ids = self - .matching_resource_ids(&db, tenant_id, contained_type, &top_query) + .matching_resource_ids( + &db, + tenant_id, + contained_type, + &top_query, + &missing_support, + ) .await?; Some(self.build_resource_filter( tenant_id, contained_type, &top_query, + &missing_support, matched_ids.as_ref(), None, )?) @@ -2692,6 +2801,7 @@ impl MongoBackend { &self, tenant: &TenantContext, query: &SearchQuery, + missing_support: &MissingSupport, db: &mongodb::Database, tenant_id: &str, matched_ids: Option>, @@ -2707,12 +2817,26 @@ impl MongoBackend { let allowed: Option> = match matched_ids { Some(set) if set.is_empty() => Some(set), Some(set) => Some( - self.resource_level_ids(db, tenant_id, &query.resource_type, query, Some(&set)) - .await?, + self.resource_level_ids( + db, + tenant_id, + &query.resource_type, + query, + missing_support, + Some(&set), + ) + .await?, ), - None if self.has_resource_level_params(tenant_id, query) => Some( - self.resource_level_ids(db, tenant_id, &query.resource_type, query, None) - .await?, + None if has_resource_level_params(query, missing_support) => Some( + self.resource_level_ids( + db, + tenant_id, + &query.resource_type, + query, + missing_support, + None, + ) + .await?, ), None => None, }; @@ -2750,6 +2874,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + missing_support, Some(&id_set), None, )?; @@ -2824,15 +2949,6 @@ impl MongoBackend { }) } - /// True when the query carries a predicate that lives on the resource - /// document rather than in the search index. - fn has_resource_level_params(&self, tenant_id: &str, query: &SearchQuery) -> bool { - query.parameters.iter().any(|p| { - matches!(p.name.as_str(), "_id" | "_lastUpdated") - || self.is_missing_resource_parameter(tenant_id, &query.resource_type, p) - }) - } - /// Ids of the live resources that satisfy the resource-level predicates /// (`_id`, `_lastUpdated`) within `matched_ids` — also the liveness /// filter for every filtered parameter sort. This is the same predicate @@ -2844,10 +2960,17 @@ impl MongoBackend { tenant_id: &str, resource_type: &str, query: &SearchQuery, + missing_support: &MissingSupport, matched_ids: Option<&HashSet>, ) -> StorageResult> { - let filter = - self.build_resource_filter(tenant_id, resource_type, query, matched_ids, None)?; + let filter = self.build_resource_filter( + tenant_id, + resource_type, + query, + missing_support, + matched_ids, + None, + )?; let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); let cursor = resources .find(filter) @@ -3065,6 +3188,7 @@ impl MongoBackend { tenant_id: &str, resource_type: &str, query: &SearchQuery, + missing_support: &MissingSupport, ) -> StorageResult>> { let search_index = db.collection::(MongoBackend::SEARCH_INDEX_COLLECTION); @@ -3209,6 +3333,7 @@ impl MongoBackend { } return self .seeded_matching_resource_ids( + db, &search_index, tenant_id, resource_type, @@ -3216,6 +3341,7 @@ impl MongoBackend { &HashMap::new(), &missing, ¬_params, + missing_support, compartment, seed, ) @@ -3230,6 +3356,7 @@ impl MongoBackend { resource_type, &missing, ¬_params, + missing_support, query, ) .await; @@ -3414,6 +3541,7 @@ impl MongoBackend { let seed = id_seed.expect("Driver::Seed implies a positive _id seed"); return self .seeded_matching_resource_ids( + db, &search_index, tenant_id, resource_type, @@ -3421,6 +3549,7 @@ impl MongoBackend { &identifier_filters, &missing, ¬_params, + missing_support, compartment, seed, ) @@ -3480,6 +3609,7 @@ impl MongoBackend { let candidates = self .retain_matching_candidates( + db, &search_index, tenant_id, resource_type, @@ -3487,6 +3617,7 @@ impl MongoBackend { &identifier_filters, &missing, ¬_params, + missing_support, compartment, Some(driver_idx), candidates, @@ -3517,6 +3648,7 @@ impl MongoBackend { #[allow(clippy::too_many_arguments)] async fn retain_matching_candidates( &self, + db: &mongodb::Database, search_index: &mongodb::Collection, tenant_id: &str, resource_type: &str, @@ -3524,6 +3656,7 @@ impl MongoBackend { identifier_filters: &HashMap, missing: &[&SearchParameter], not_params: &[&SearchParameter], + missing_support: &MissingSupport, compartment: Option<&CompartmentMembership>, skip_idx: Option, mut candidates: HashSet, @@ -3575,10 +3708,9 @@ impl MongoBackend { if candidates.is_empty() { break; } - if self.is_missing_resource_parameter(tenant_id, resource_type, param) { - let db = self.get_database().await?; + if missing_support.is_resource_parameter(param) { candidates = self - .metadata_missing_ids(&db, tenant_id, resource_type, param, Some(&candidates)) + .metadata_missing_ids(db, tenant_id, resource_type, param, Some(&candidates)) .await?; continue; } @@ -3643,6 +3775,7 @@ impl MongoBackend { #[allow(clippy::too_many_arguments)] async fn seeded_matching_resource_ids( &self, + db: &mongodb::Database, search_index: &mongodb::Collection, tenant_id: &str, resource_type: &str, @@ -3650,6 +3783,7 @@ impl MongoBackend { identifier_filters: &HashMap, missing: &[&SearchParameter], not_params: &[&SearchParameter], + missing_support: &MissingSupport, compartment: Option<&CompartmentMembership>, seed: HashSet, ) -> StorageResult> { @@ -3661,6 +3795,7 @@ impl MongoBackend { let candidates: HashSet = chunk.iter().map(|id| (*id).clone()).collect(); let candidates = self .retain_matching_candidates( + db, search_index, tenant_id, resource_type, @@ -3668,6 +3803,7 @@ impl MongoBackend { identifier_filters, missing, not_params, + missing_support, compartment, None, candidates, @@ -3696,6 +3832,7 @@ impl MongoBackend { resource_type: &str, missing: &[&SearchParameter], not_params: &[&SearchParameter], + missing_support: &MissingSupport, query: &SearchQuery, ) -> StorageResult>> { let has_compartment = query @@ -3705,7 +3842,7 @@ impl MongoBackend { let mut matched: Option> = None; for param in missing { - if self.is_missing_resource_parameter(tenant_id, resource_type, param) { + if missing_support.is_resource_parameter(param) { let ids = self .metadata_missing_ids(db, tenant_id, resource_type, param, matched.as_ref()) .await?; @@ -5107,6 +5244,7 @@ impl MongoBackend { tenant_id: &str, resource_type: &str, query: &SearchQuery, + missing_support: &MissingSupport, matched_ids: Option<&HashSet>, cursor: Option<(&PageCursor, &CursorKeyset)>, ) -> StorageResult { @@ -5130,7 +5268,7 @@ impl MongoBackend { } for param in &query.parameters { - if self.is_missing_resource_parameter(tenant_id, resource_type, param) { + if missing_support.is_resource_parameter(param) { conditions.push(missing_resource_condition(¶m.name)); continue; } @@ -6715,7 +6853,7 @@ mod value_list_tests { }); let filter = backend - .build_resource_filter("t1", "Patient", &query, None, None) + .build_resource_filter("t1", "Patient", &query, &MissingSupport::none(), None, None) .expect("valid filter"); let and_arms = filter.get_array("$and").expect("$and array"); @@ -6759,7 +6897,7 @@ mod value_list_tests { }); let filter = backend - .build_resource_filter("t1", "Patient", &query, None, None) + .build_resource_filter("t1", "Patient", &query, &MissingSupport::none(), None, None) .expect("valid filter"); let and_arms = filter.get_array("$and").expect("$and array"); @@ -7358,7 +7496,7 @@ mod metadata_param_modifier_tests { fn condition(backend: &MongoBackend, param: SearchParameter) -> Document { let query = SearchQuery::new("Patient").with_parameter(param); let filter = backend - .build_resource_filter("t", "Patient", &query, None, None) + .build_resource_filter("t", "Patient", &query, &MissingSupport::none(), None, None) .expect("filter should build"); filter.get_array("$and").expect("$and array")[1] .as_document() diff --git a/crates/persistence/src/backends/mongodb/search_impl/search_plan.rs b/crates/persistence/src/backends/mongodb/search_impl/search_plan.rs index 216029119a..bcd84b5026 100644 --- a/crates/persistence/src/backends/mongodb/search_impl/search_plan.rs +++ b/crates/persistence/src/backends/mongodb/search_impl/search_plan.rs @@ -752,6 +752,7 @@ impl MongoBackend { db: &mongodb::Database, tenant_id: &str, query: &SearchQuery, + missing_support: &super::MissingSupport, purpose: SearchFilterPurpose, page_policy: NoTotalPagePolicy, index_cursor_filter: Option<&Document>, @@ -773,9 +774,10 @@ impl MongoBackend { { return Ok(None); } - let missing_resource_filter = query.parameters.iter().any(|param| { - self.is_missing_resource_parameter(tenant_id, &query.resource_type, param) - }); + let missing_resource_filter = query + .parameters + .iter() + .any(|param| missing_support.is_resource_parameter(param)); let indexed: Vec<_> = query .parameters .iter() @@ -799,11 +801,7 @@ impl MongoBackend { matches!(param.modifier, Some(SearchModifier::Not)) || (matches!(param.modifier, Some(SearchModifier::Missing)) && (param.param_type == SearchParamType::Composite - || !self.supports_indexed_missing( - tenant_id, - &query.resource_type, - param, - ))) + || !missing_support.supports(param))) }); if !has_complement { let probe_timeout = self.probe_timeout(); @@ -865,9 +863,9 @@ impl MongoBackend { && self.search_indexes_ready() && missing_resource_filter && matches!(query.sort.as_slice(), [sort] if sort.parameter == "_id") - && indexed.iter().any(|param| { - !self.is_missing_resource_parameter(tenant_id, &query.resource_type, param) - }); + && indexed + .iter() + .any(|param| !missing_support.is_resource_parameter(param)); let batch_page = timestamp_batch_page || id_batch_page; // Keep selective positive predicates on the index-first path. A // bounded count transfers only a scalar; broad searches never collect @@ -875,12 +873,12 @@ impl MongoBackend { let search_index = db.collection::(MongoBackend::SEARCH_INDEX_COLLECTION); let mut driver = None; for (position, param) in indexed.iter().enumerate() { - if self.is_missing_resource_parameter(tenant_id, &query.resource_type, param) { + if missing_support.is_resource_parameter(param) { continue; } if matches!(param.modifier, Some(SearchModifier::Missing)) && (param.param_type == SearchParamType::Composite - || !self.supports_indexed_missing(tenant_id, &query.resource_type, param)) + || !missing_support.supports(param)) { // Presence is a positive predicate: drive from its typed partial // index, rather than enumerating resources with no value. @@ -928,7 +926,7 @@ impl MongoBackend { } if matches!(purpose, SearchFilterPurpose::Count) && let Some(plan) = self - .missing_count_plan(db, tenant_id, query, &indexed, facts) + .missing_count_plan(db, tenant_id, query, missing_support, &indexed, facts) .await? { return Ok(Some(SearchFilterPlan::BatchedIndex(plan))); @@ -938,7 +936,7 @@ impl MongoBackend { let mut missing = Vec::new(); let mut stages = Vec::new(); for (position, param) in indexed.into_iter().enumerate() { - if self.is_missing_resource_parameter(tenant_id, &query.resource_type, param) { + if missing_support.is_resource_parameter(param) { let presence = missing_presence_filter(tenant_id, &query.resource_type, param); if batch_page { missing.push(presence.clone()); @@ -1060,6 +1058,7 @@ impl MongoBackend { db: &mongodb::Database, tenant_id: &str, query: &SearchQuery, + missing_support: &super::MissingSupport, indexed: &[&SearchParameter], facts: &mut ProbeFacts, ) -> StorageResult> { @@ -1067,7 +1066,7 @@ impl MongoBackend { let mut positive = Vec::new(); let mut missing = Vec::new(); for param in indexed { - if self.is_missing_resource_parameter(tenant_id, resource_type, param) { + if missing_support.is_resource_parameter(param) { missing.push(missing_presence_filter(tenant_id, resource_type, param)); } else if matches!(param.modifier, Some(SearchModifier::Not)) || matches!(param.modifier, Some(SearchModifier::Missing)) diff --git a/crates/persistence/src/backends/mongodb/search_impl/search_plan/execution_tests.rs b/crates/persistence/src/backends/mongodb/search_impl/search_plan/execution_tests.rs index 24e1ea42ab..0343d16bf7 100644 --- a/crates/persistence/src/backends/mongodb/search_impl/search_plan/execution_tests.rs +++ b/crates/persistence/src/backends/mongodb/search_impl/search_plan/execution_tests.rs @@ -805,6 +805,9 @@ async fn stale_probe_changes_only_driver_order_and_refresh_clears_it() { &fixture.db, PLAN_TENANT, &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -852,6 +855,9 @@ async fn stale_probe_changes_only_driver_order_and_refresh_clears_it() { &fixture.db, PLAN_TENANT, &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -1155,6 +1161,7 @@ async fn configured_probe_budget_reaches_the_plan_that_probes() { &fixture.db, PLAN_TENANT, &query, + &configured.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -1534,6 +1541,9 @@ async fn request_reuses_untimed_selective_probe_across_planning_calls() { &fixture.db, PLAN_TENANT, &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -2185,6 +2195,9 @@ async fn missing_metadata_selection_preserves_other_candidate_routes() { &fixture.db, PLAN_TENANT, &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -2198,7 +2211,15 @@ async fn missing_metadata_selection_preserves_other_candidate_routes() { )); let (selected, ids) = fixture .backend - .plan_or_matching_ids(&fixture.db, PLAN_TENANT, &query, plan) + .plan_or_matching_ids( + &fixture.db, + PLAN_TENANT, + &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), + plan, + ) .await .unwrap(); assert!( @@ -2226,7 +2247,12 @@ async fn missing_metadata_selection_preserves_other_candidate_routes() { assert!( fixture .backend - .is_missing_resource_parameter(PLAN_TENANT, "Observation", &absent) + .classify_missing( + PLAN_TENANT, + "Observation", + &SearchQuery::new("Observation").with_parameter(absent.clone()), + ) + .is_resource_parameter(&absent) ); let mut subject = parameter("subject", "urn:selector-test|unknown"); subject.param_type = SearchParamType::Reference; @@ -2240,6 +2266,9 @@ async fn missing_metadata_selection_preserves_other_candidate_routes() { &fixture.db, PLAN_TENANT, &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -2253,7 +2282,15 @@ async fn missing_metadata_selection_preserves_other_candidate_routes() { ); let (selected, ids) = fixture .backend - .plan_or_matching_ids(&fixture.db, PLAN_TENANT, &query, plan) + .plan_or_matching_ids( + &fixture.db, + PLAN_TENANT, + &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), + plan, + ) .await .unwrap(); assert!( diff --git a/crates/persistence/src/backends/mongodb/search_impl/search_plan/legacy_candidates_tests.rs b/crates/persistence/src/backends/mongodb/search_impl/search_plan/legacy_candidates_tests.rs index e6a806565b..8b00782040 100644 --- a/crates/persistence/src/backends/mongodb/search_impl/search_plan/legacy_candidates_tests.rs +++ b/crates/persistence/src/backends/mongodb/search_impl/search_plan/legacy_candidates_tests.rs @@ -1,6 +1,7 @@ //! Access paths and result parity for the bounded legacy candidate collector. use super::super::super::CANDIDATE_BATCH_SIZE; +use super::super::super::MissingSupport; use super::super::super::search_access::SEARCH_TOKEN_MEMBERSHIP_INDEX; use super::*; use crate::backends::mongodb::schema::RESOURCES_TYPE_SCAN_INDEX; @@ -457,7 +458,13 @@ async fn legacy_date_discovery_retries_only_a_missing_date_index() { for query in &queries { let actual = fixture .backend - .matching_resource_ids(&fixture.db, PLAN_TENANT, "Observation", query) + .matching_resource_ids( + &fixture.db, + PLAN_TENANT, + "Observation", + query, + &MissingSupport::none(), + ) .await .unwrap(); assert_eq!(actual, Some(expected.clone())); @@ -498,7 +505,13 @@ async fn legacy_discovery_errors_and_required_membership_index_errors_propagate( .await; let result = fixture .backend - .matching_resource_ids(&fixture.db, PLAN_TENANT, "Observation", query) + .matching_resource_ids( + &fixture.db, + PLAN_TENANT, + "Observation", + query, + &MissingSupport::none(), + ) .await; let after = disable_failpoint(&session).await; assert!( @@ -517,7 +530,13 @@ async fn legacy_discovery_errors_and_required_membership_index_errors_propagate( profile_start(&fixture.db).await; let error = fixture .backend - .matching_resource_ids(&fixture.db, PLAN_TENANT, "Observation", &pair) + .matching_resource_ids( + &fixture.db, + PLAN_TENANT, + "Observation", + &pair, + &MissingSupport::none(), + ) .await .unwrap_err(); assert!( @@ -642,6 +661,7 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() let actual = fixture .backend .retain_matching_candidates( + &fixture.db, &index, PLAN_TENANT, "Observation", @@ -649,6 +669,7 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() &HashMap::new(), &[], &[], + &MissingSupport::none(), None, skip, candidates.clone(), @@ -699,6 +720,7 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() let absent = fixture .backend .retain_matching_candidates( + &fixture.db, &index, PLAN_TENANT, "Observation", @@ -706,6 +728,11 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() &HashMap::new(), &[&missing], &[], + &fixture.backend.classify_missing( + PLAN_TENANT, + "Observation", + &SearchQuery::new("Observation").with_parameter(missing.clone()), + ), None, skip, candidates.clone(), @@ -722,6 +749,7 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() let not_red = fixture .backend .retain_matching_candidates( + &fixture.db, &index, PLAN_TENANT, "Observation", @@ -729,6 +757,7 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() &HashMap::new(), &[], &[&excluded], + &MissingSupport::none(), None, skip, candidates.clone(), @@ -745,6 +774,7 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() let in_compartment = fixture .backend .retain_matching_candidates( + &fixture.db, &index, PLAN_TENANT, "Observation", @@ -752,6 +782,7 @@ async fn bounded_membership_preserves_types_multivalues_and_candidate_origins() &HashMap::new(), &[], &[], + &MissingSupport::none(), Some(&compartment), skip, candidates.clone(), @@ -805,6 +836,7 @@ async fn legacy_token_membership_shares_named_system_fallback() { let actual = fixture .backend .retain_matching_candidates( + &fixture.db, &index, PLAN_TENANT, "Observation", @@ -812,6 +844,7 @@ async fn legacy_token_membership_shares_named_system_fallback() { &HashMap::new(), &[], &[], + &MissingSupport::none(), None, skip, candidates.clone(), @@ -890,6 +923,7 @@ async fn membership_intersects_occurrences_without_requiring_one_matching_value( let actual = fixture .backend .retain_matching_candidates( + &fixture.db, &index, PLAN_TENANT, "Observation", @@ -897,6 +931,7 @@ async fn membership_intersects_occurrences_without_requiring_one_matching_value( &HashMap::new(), &[], &[], + &MissingSupport::none(), None, skip, initial.clone(), diff --git a/crates/persistence/src/backends/mongodb/search_impl/search_plan/resource_page_tests.rs b/crates/persistence/src/backends/mongodb/search_impl/search_plan/resource_page_tests.rs index 410a454c62..2d69b2dfe1 100644 --- a/crates/persistence/src/backends/mongodb/search_impl/search_plan/resource_page_tests.rs +++ b/crates/persistence/src/backends/mongodb/search_impl/search_plan/resource_page_tests.rs @@ -199,6 +199,11 @@ async fn missing_membership_work_is_bounded_by_candidate_ids() { db, PLAN_TENANT, &query, + &fixture.backend.classify_missing( + PLAN_TENANT, + &query.resource_type, + &query, + ), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -263,6 +268,11 @@ async fn missing_membership_work_is_bounded_by_candidate_ids() { db, PLAN_TENANT, &query, + &fixture.backend.classify_missing( + PLAN_TENANT, + &query.resource_type, + &query + ), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -451,6 +461,7 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -463,7 +474,14 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { }; assert_eq!(facts.reuses, 1); let filter = backend - .build_resource_filter(PLAN_TENANT, "Observation", &query, None, None) + .build_resource_filter( + PLAN_TENANT, + "Observation", + &query, + &backend.classify_missing(PLAN_TENANT, "Observation", &query), + None, + None, + ) .unwrap(); let actual = plan .execute_page(db, filter, None, doc! { "id": 1 }, 0, 21) @@ -487,6 +505,7 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), purpose, NoTotalPagePolicy::Streaming, None, @@ -515,7 +534,13 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { "{code}: unexpected planning probes" ); let (selected, ids) = backend - .plan_or_matching_ids(db, PLAN_TENANT, &query, plan) + .plan_or_matching_ids( + db, + PLAN_TENANT, + &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), + plan, + ) .await .unwrap(); if small { @@ -555,6 +580,7 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -577,7 +603,13 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { "pure missing pages must not acquire a probe" ); let (selected, ids) = backend - .plan_or_matching_ids(db, PLAN_TENANT, &query, plan) + .plan_or_matching_ids( + db, + PLAN_TENANT, + &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), + plan, + ) .await .unwrap(); assert!(matches!( @@ -606,6 +638,7 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -733,6 +766,7 @@ async fn missing_id_cursor_candidates_bound_probes_reads_and_cache_keys() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, Some(&boundary), @@ -767,6 +801,7 @@ async fn missing_id_cursor_candidates_bound_probes_reads_and_cache_keys() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, Some(&boundary), @@ -1030,6 +1065,11 @@ async fn missing_id_batches_navigate_fifty_pages_with_independent_date_bounds() db, PLAN_TENANT, &query, + &fixture.backend.classify_missing( + PLAN_TENANT, + &query.resource_type, + &query + ), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -1045,6 +1085,9 @@ async fn missing_id_batches_navigate_fifty_pages_with_independent_date_bounds() db, PLAN_TENANT, &query, + &fixture + .backend + .classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -1569,6 +1612,11 @@ async fn timestamp_pages_filter_before_offsets_and_preserve_cursor_totals() { &data.fixture.db, PLAN_TENANT, &query, + &data.fixture.backend.classify_missing( + PLAN_TENANT, + &query.resource_type, + &query + ), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -2152,6 +2200,11 @@ async fn complement_pages_batch_exclusions_and_preserve_absence_and_cursors() { db, PLAN_TENANT, &query, + &data.fixture.backend.classify_missing( + PLAN_TENANT, + &query.resource_type, + &query, + ), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -2312,6 +2365,7 @@ async fn complement_planning_preserves_selective_and_unknown_probe_routes() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -2349,6 +2403,7 @@ async fn complement_planning_preserves_selective_and_unknown_probe_routes() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Count, NoTotalPagePolicy::Streaming, None, @@ -2376,6 +2431,7 @@ async fn complement_planning_preserves_selective_and_unknown_probe_routes() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -2452,6 +2508,7 @@ async fn complement_date_pages_reuse_metadata_without_hinting_presence() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -2493,6 +2550,7 @@ async fn complement_date_pages_reuse_metadata_without_hinting_presence() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, @@ -2530,6 +2588,7 @@ async fn complement_date_pages_reuse_metadata_without_hinting_presence() { db, PLAN_TENANT, &query, + &backend.classify_missing(PLAN_TENANT, &query.resource_type, &query), SearchFilterPurpose::Page, NoTotalPagePolicy::Streaming, None, diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index 86bb51ff94..4eae2b0179 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -49,6 +49,7 @@ use super::deferred_index::{ }; use super::retry::{BUNDLE_TRANSACTION_RETRY, WRITE_ABORT_RETRY, jitter_fraction}; use super::schema::{RESOURCES_IDENTITY_INDEX, RESOURCES_TYPE_SCAN_INDEX}; +use super::search_impl::MissingSupport; pub(super) fn internal_error(message: String) -> StorageError { StorageError::Backend(BackendError::Internal { @@ -5968,8 +5969,14 @@ impl MongoBackend { ..Default::default() }; let tenant_id = tenant.tenant_id().as_str(); - let filter = - self.build_resource_filter(tenant_id, resource_type, &query, None, None)?; + let filter = self.build_resource_filter( + tenant_id, + resource_type, + &query, + &MissingSupport::none(), + None, + None, + )?; let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); let cursor = resources .find(filter) @@ -6180,6 +6187,7 @@ impl MongoBackend { tenant_id, resource_type, &resource_query, + &MissingSupport::none(), None, None, )?; @@ -6295,6 +6303,7 @@ impl MongoBackend { tenant_id, resource_type, &query, + &MissingSupport::none(), None, None, )?);