From 4fa437859eb3e53b6f81662fa12bc0a8302ced38 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Wed, 21 Oct 2026 13:24:50 -0400 Subject: [PATCH] perf(persistence/mongodb): classify metadata-backed :missing parameters once per query Whether a :missing parameter is answered from missing_params was recomputed at every planning and execution stage, each time taking the tenant registry read lock and building an extractor. A registry reload mid-query could also make the planner and executor disagree. Classify once into MissingSupport (keyed by name and type, which is all the check depended on) and pass it, with the database handle, through planning and execution. retain_matching_candidates no longer calls get_database() per metadata-backed missing parameter, and the search total reuses the page's handle and classification. Closes #2007 --- .../src/backends/mongodb/search_impl.rs | 272 +++++++++++++----- .../mongodb/search_impl/search_plan.rs | 26 +- .../search_plan/execution_tests.rs | 43 ++- .../search_plan/resource_page_tests.rs | 65 ++++- .../src/backends/mongodb/storage.rs | 13 +- 5 files changed, 326 insertions(+), 93 deletions(-) diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index 25a3ed2acf..b976373d4d 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -1037,6 +1037,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(|_| { @@ -1078,7 +1079,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( @@ -1095,6 +1103,7 @@ impl MongoBackend { &db, tenant_id, query, + &missing_support, execution.filter_purpose, execution.page_policy, index_cursor_filter.as_ref(), @@ -1102,7 +1111,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", @@ -1153,6 +1162,7 @@ impl MongoBackend { .search_param_sorted( tenant, query, + &missing_support, &db, tenant_id, matched_ids, @@ -1169,6 +1179,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + &missing_support, matched_ids.as_ref(), None, )?; @@ -1193,6 +1204,7 @@ impl MongoBackend { &db, tenant_id, query, + &missing_support, plan, scope_filter, cursor_condition, @@ -1206,7 +1218,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() }; @@ -1334,7 +1353,9 @@ impl MongoBackend { Some( self.count_standard_with_facts( tenant, + &db, query, + &missing_support, facts.count_selection(), &mut probe_facts, ) @@ -1626,50 +1647,107 @@ fn contained_composite_value_stages( Ok(first) } +/// 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 answer depends only on a parameter's name and type, +/// which is what the set holds. +#[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 { + !matches!(param.name.as_str(), "_id" | "_lastUpdated") + && 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) - }) - } - - fn is_missing_resource_parameter( - &self, - tenant_id: &str, - resource_type: &str, - param: &SearchParameter, - ) -> bool { - !matches!(param.name.as_str(), "_id" | "_lastUpdated") - && 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(); + } + let extractor = self.tenant_extractor(tenant_id); + MissingSupport { + indexed: definitions + .into_iter() + .filter(|(_, definition)| extractor.supports_missing(resource_type, definition)) + .map(|(param, _)| (param.name.clone(), param.param_type)) + .collect(), + } } /// Chooses between `plan` and the candidate id set. `IndexIntersection` @@ -1682,6 +1760,7 @@ impl MongoBackend { db: &mongodb::Database, tenant_id: &str, query: &SearchQuery, + missing_support: &MissingSupport, plan: Option, ) -> StorageResult<(Option, Option>)> { match plan { @@ -1690,14 +1769,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; @@ -1725,6 +1805,7 @@ impl MongoBackend { db: &mongodb::Database, tenant_id: &str, query: &SearchQuery, + missing_support: &MissingSupport, plan: search_plan::ResourceBatchPlan, scope_filter: Document, cursor_condition: Option, @@ -1752,7 +1833,10 @@ impl MongoBackend { run.finish(db).await? } search_plan::ResourcePageSample::Sparse { live_matches } => { - 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) => { search_plan::record_resource_page_fallback(live_matches, id_set.len()); let filter = with_cursor_condition( @@ -1760,6 +1844,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + missing_support, Some(&id_set), None, )?, @@ -1803,9 +1888,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)), @@ -1857,8 +1943,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 @@ -1868,19 +1957,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, @@ -1888,7 +1979,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", @@ -1900,6 +1991,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + missing_support, matched_ids.as_ref(), None, )?; @@ -1911,7 +2003,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"); @@ -2074,13 +2166,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, )?) @@ -2675,6 +2776,7 @@ impl MongoBackend { &self, tenant: &TenantContext, query: &SearchQuery, + missing_support: &MissingSupport, db: &mongodb::Database, tenant_id: &str, matched_ids: Option>, @@ -2690,12 +2792,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, }; @@ -2733,6 +2849,7 @@ impl MongoBackend { tenant_id, &query.resource_type, query, + missing_support, Some(&id_set), None, )?; @@ -2807,15 +2924,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 @@ -2827,10 +2935,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) @@ -3048,6 +3163,7 @@ impl MongoBackend { tenant_id: &str, resource_type: &str, query: &SearchQuery, + missing_support: &MissingSupport, ) -> StorageResult>> { let search_index = db.collection::(MongoBackend::SEARCH_INDEX_COLLECTION); @@ -3192,6 +3308,7 @@ impl MongoBackend { } return self .seeded_matching_resource_ids( + db, &search_index, tenant_id, resource_type, @@ -3199,6 +3316,7 @@ impl MongoBackend { &HashMap::new(), &missing, ¬_params, + missing_support, compartment, seed, ) @@ -3213,6 +3331,7 @@ impl MongoBackend { resource_type, &missing, ¬_params, + missing_support, query, ) .await; @@ -3385,6 +3504,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, @@ -3392,6 +3512,7 @@ impl MongoBackend { &identifier_filters, &missing, ¬_params, + missing_support, compartment, seed, ) @@ -3445,6 +3566,7 @@ impl MongoBackend { let candidates = self .retain_matching_candidates( + db, &search_index, tenant_id, resource_type, @@ -3452,6 +3574,7 @@ impl MongoBackend { &identifier_filters, &missing, ¬_params, + missing_support, compartment, Some(driver_idx), false, @@ -3501,6 +3624,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, @@ -3508,6 +3632,7 @@ impl MongoBackend { identifier_filters: &HashMap, missing: &[&SearchParameter], not_params: &[&SearchParameter], + missing_support: &MissingSupport, compartment: Option<&CompartmentMembership>, skip_idx: Option, hint: bool, @@ -3595,10 +3720,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; } @@ -3780,6 +3904,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, @@ -3787,6 +3912,7 @@ impl MongoBackend { identifier_filters: &HashMap, missing: &[&SearchParameter], not_params: &[&SearchParameter], + missing_support: &MissingSupport, compartment: Option<&CompartmentMembership>, seed: HashSet, ) -> StorageResult> { @@ -3798,6 +3924,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, @@ -3805,6 +3932,7 @@ impl MongoBackend { identifier_filters, missing, not_params, + missing_support, compartment, None, true, @@ -3834,6 +3962,7 @@ impl MongoBackend { resource_type: &str, missing: &[&SearchParameter], not_params: &[&SearchParameter], + missing_support: &MissingSupport, query: &SearchQuery, ) -> StorageResult>> { let has_compartment = query @@ -3843,7 +3972,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?; @@ -5248,6 +5377,7 @@ impl MongoBackend { tenant_id: &str, resource_type: &str, query: &SearchQuery, + missing_support: &MissingSupport, matched_ids: Option<&HashSet>, cursor: Option<(&PageCursor, &CursorKeyset)>, ) -> StorageResult { @@ -5271,7 +5401,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; } @@ -6856,7 +6986,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"); @@ -6900,7 +7030,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"); @@ -7499,7 +7629,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 89b1c0c52b..e7adcf7cfb 100644 --- a/crates/persistence/src/backends/mongodb/search_impl/search_plan.rs +++ b/crates/persistence/src/backends/mongodb/search_impl/search_plan.rs @@ -858,6 +858,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>, @@ -879,9 +880,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() @@ -905,11 +907,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(); @@ -971,9 +969,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 @@ -981,12 +979,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. @@ -1037,7 +1035,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()); 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 c5a1dd5e82..faa3a7c692 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 @@ -778,6 +778,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, @@ -825,6 +828,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, @@ -1128,6 +1134,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, @@ -1507,6 +1514,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, @@ -2158,6 +2168,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, @@ -2171,7 +2184,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!( @@ -2191,7 +2212,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; @@ -2205,6 +2231,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, @@ -2218,7 +2247,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/resource_page_tests.rs b/crates/persistence/src/backends/mongodb/search_impl/search_plan/resource_page_tests.rs index 3c1fcb890d..0ae6deab8c 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 @@ -198,6 +198,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, @@ -262,6 +267,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, @@ -452,6 +462,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, @@ -464,7 +475,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) @@ -488,6 +506,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, @@ -511,7 +530,13 @@ async fn missing_id_pages_preserve_small_count_and_broad_fallbacks() { "small and count routes must not add a bounded 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(); if small { @@ -551,6 +576,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, @@ -573,7 +599,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!( @@ -602,6 +634,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, @@ -728,6 +761,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), @@ -762,6 +796,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), @@ -1027,6 +1062,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, @@ -1042,6 +1082,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, @@ -1557,6 +1600,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, @@ -2085,6 +2133,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, @@ -2245,6 +2298,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, @@ -2282,6 +2336,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, @@ -2309,6 +2364,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, @@ -2384,6 +2440,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, @@ -2425,6 +2482,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, @@ -2462,6 +2520,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 a614311c3a..48eb472df9 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 { @@ -5906,8 +5907,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) @@ -6118,6 +6125,7 @@ impl MongoBackend { tenant_id, resource_type, &resource_query, + &MissingSupport::none(), None, None, )?; @@ -6233,6 +6241,7 @@ impl MongoBackend { tenant_id, resource_type, &query, + &MissingSupport::none(), None, None, )?);