From fc10cac5909b232c3ed02720237cc7df5c3c88bb Mon Sep 17 00:00:00 2001 From: trinity-1686a Date: Tue, 7 Jul 2026 12:02:52 +0200 Subject: [PATCH] enforce retention at query time --- .../src/actors/delete_task_planner.rs | 1 + .../protos/quickwit/search.proto | 4 + .../src/codegen/quickwit/quickwit.search.rs | 4 + .../quickwit-search/src/cluster_client.rs | 1 + quickwit/quickwit-search/src/leaf.rs | 77 +++++++++- quickwit/quickwit-search/src/retry/search.rs | 1 + quickwit/quickwit-search/src/root.rs | 134 +++++++++++++++++- 7 files changed, 217 insertions(+), 5 deletions(-) diff --git a/quickwit/quickwit-janitor/src/actors/delete_task_planner.rs b/quickwit/quickwit-janitor/src/actors/delete_task_planner.rs index bf42e949c36..aba6aca77c6 100644 --- a/quickwit/quickwit-janitor/src/actors/delete_task_planner.rs +++ b/quickwit/quickwit-janitor/src/actors/delete_task_planner.rs @@ -323,6 +323,7 @@ impl DeleteTaskPlanner { IndexMetasForLeafSearch { doc_mapper_str: doc_mapper_str.to_string(), index_uri, + retention_timestamp_cutoff_opt: None, }, ); let leaf_search_request = jobs_to_leaf_request( diff --git a/quickwit/quickwit-proto/protos/quickwit/search.proto b/quickwit/quickwit-proto/protos/quickwit/search.proto index 4d060d6e1be..cd88e7d25d8 100644 --- a/quickwit/quickwit-proto/protos/quickwit/search.proto +++ b/quickwit/quickwit-proto/protos/quickwit/search.proto @@ -517,6 +517,10 @@ message LeafRequestRef { // Index split ids to apply the query on. // This ids are resolved from the index_uri defined in the search_request. repeated SplitIdAndFooterOffsets split_offsets = 3; + + // Unix timestamp in seconds. When set, only documents with a timestamp >= + // this value are considered. + optional int64 retention_timestamp_cutoff = 4; } message SplitIdAndFooterOffsets { diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs index a9109c1ad11..b409891ad10 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs @@ -473,6 +473,10 @@ pub struct LeafRequestRef { /// This ids are resolved from the index_uri defined in the search_request. #[prost(message, repeated, tag = "3")] pub split_offsets: ::prost::alloc::vec::Vec, + /// Unix timestamp in seconds. When set, only documents with a timestamp >= + /// this value are considered. + #[prost(int64, optional, tag = "4")] + pub retention_timestamp_cutoff: ::core::option::Option, } #[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)] #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] diff --git a/quickwit/quickwit-search/src/cluster_client.rs b/quickwit/quickwit-search/src/cluster_client.rs index 168ab5993b0..1c9dd7fe0a4 100644 --- a/quickwit/quickwit-search/src/cluster_client.rs +++ b/quickwit/quickwit-search/src/cluster_client.rs @@ -373,6 +373,7 @@ mod tests { num_docs: 0, }, ], + retention_timestamp_cutoff: None, }], } } diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index 121e2d7df02..83bfa456733 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -1491,6 +1491,33 @@ impl CanSplitDoBetter { } } +/// Returns a clone of `search_request` with `start_timestamp` tightened to the retention cutoff. +/// +/// When an index has a retention policy the root sets `retention_timestamp_cutoff` on the +/// `LeafRequestRef`. We apply it here as an updated `start_timestamp` so that +/// `remove_redundant_timestamp_range` injects the appropriate Tantivy range filter on any split +/// that straddles the retention boundary. The user-supplied `start_timestamp` is respected: we +/// take the max of the two values so the more restrictive bound always wins. +fn apply_retention_cutoff( + search_request: Arc, + cutoff: Option, +) -> Arc { + let Some(cutoff) = cutoff else { + return search_request; + }; + let effective_start = match search_request.start_timestamp { + Some(existing) => existing.max(cutoff), + None => cutoff, + }; + if search_request.start_timestamp != Some(effective_start) { + let mut req = (*search_request).clone(); + req.start_timestamp = Some(effective_start); + Arc::new(req) + } else { + search_request + } +} + /// Searches multiple splits, potentially in multiple indexes, sitting on different storages and /// having different doc mappings. #[instrument(skip_all, fields(index = ?leaf_search_request.search_request.as_ref().unwrap().index_id_patterns))] @@ -1543,14 +1570,17 @@ pub async fn multi_index_leaf_search( let storage_resolver = storage_resolver.clone(); let searcher_context = searcher_context.clone(); - let search_request = search_request.clone(); + let search_request_for_index = apply_retention_cutoff( + search_request.clone(), + leaf_search_request_ref.retention_timestamp_cutoff, + ); leaf_request_futures.spawn({ async move { let storage = storage_resolver.resolve(&index_uri).await?; single_doc_mapping_leaf_search( searcher_context, - search_request, + search_request_for_index, storage, leaf_search_request_ref.split_offsets, doc_mapper, @@ -1711,7 +1741,7 @@ async fn run_offloaded_search_tasks( let batch_split_ids: Vec = batch.iter().map(|split| split.split_id.clone()).collect(); let leaf_request = LeafSearchRequest { - // Note this is not the split-specific rewritten request, we ship the main request, + // Note this is not the split-specific rewritten request, we ship the index-specific request, // and the leaf will apply the split specific rewrite on its own. search_request: Some(search_request.clone()), doc_mappers: vec![doc_mapper_str.clone()], @@ -1720,6 +1750,7 @@ async fn run_offloaded_search_tasks( index_uri_ord: 0, doc_mapper_ord: 0, split_offsets: batch, + retention_timestamp_cutoff: None, }], }; let invoker = lambda_invoker.clone(); @@ -3116,4 +3147,44 @@ mod tests { } } } + + #[test] + fn test_retention_cutoff_no_existing_start_timestamp() { + let request = Arc::new(SearchRequest { + start_timestamp: None, + ..Default::default() + }); + let result = super::apply_retention_cutoff(request, Some(1000)); + assert_eq!(result.start_timestamp, Some(1000)); + } + + #[test] + fn test_retention_cutoff_more_restrictive_than_user_start() { + let request = Arc::new(SearchRequest { + start_timestamp: Some(500), + ..Default::default() + }); + let result = super::apply_retention_cutoff(request, Some(1000)); + assert_eq!(result.start_timestamp, Some(1000)); + } + + #[test] + fn test_retention_cutoff_less_restrictive_than_user_start() { + let request = Arc::new(SearchRequest { + start_timestamp: Some(2000), + ..Default::default() + }); + let result = super::apply_retention_cutoff(request, Some(1000)); + assert_eq!(result.start_timestamp, Some(2000)); + } + + #[test] + fn test_retention_cutoff_none_leaves_request_unchanged() { + let request = Arc::new(SearchRequest { + start_timestamp: Some(500), + ..Default::default() + }); + let result = super::apply_retention_cutoff(request, None); + assert_eq!(result.start_timestamp, Some(500)); + } } diff --git a/quickwit/quickwit-search/src/retry/search.rs b/quickwit/quickwit-search/src/retry/search.rs index 696a352de94..d92a4e098de 100644 --- a/quickwit/quickwit-search/src/retry/search.rs +++ b/quickwit/quickwit-search/src/retry/search.rs @@ -103,6 +103,7 @@ mod tests { num_docs: 0, }, ], + retention_timestamp_cutoff: None, }], } } diff --git a/quickwit/quickwit-search/src/root.rs b/quickwit/quickwit-search/src/root.rs index f04bb437851..c759c83bbab 100644 --- a/quickwit/quickwit-search/src/root.rs +++ b/quickwit/quickwit-search/src/root.rs @@ -159,6 +159,10 @@ pub struct IndexMetasForLeafSearch { pub index_uri: Uri, /// Doc mapper json string. pub doc_mapper_str: String, + /// Retention cutoff as a Unix timestamp in seconds. Document with timestamp + /// before this value are filtered out. + #[serde(default)] + pub retention_timestamp_cutoff_opt: Option, } pub(crate) type IndexesMetasForLeafSearch = HashMap; @@ -184,6 +188,7 @@ struct RequestMetadata { fn validate_request_and_build_metadata( indexes_metadata: &[IndexMetadata], search_request: &SearchRequest, + now_secs: i64, ) -> crate::Result { validate_sort_by_fields_and_search_after( &search_request.sort_fields, @@ -258,11 +263,18 @@ fn validate_request_and_build_metadata( None, )?; + let retention_timestamp_cutoff_opt: Option = index_metadata + .index_config + .retention_policy_opt + .as_ref() + .and_then(|policy| policy.retention_period().ok()) + .map(|duration| now_secs.saturating_sub(duration.as_secs() as i64)); let index_metadata_for_leaf_search = IndexMetasForLeafSearch { index_uri: index_metadata.index_uri().clone(), doc_mapper_str: serde_json::to_string(&doc_mapper).map_err(|err| { SearchError::Internal(format!("failed to serialize doc mapper. cause: {err}")) })?, + retention_timestamp_cutoff_opt, }; indexes_meta_for_leaf_search.insert( index_metadata.index_uid.clone(), @@ -1214,6 +1226,7 @@ async fn refine_and_list_matches( query_ast_resolved: QueryAst, sort_fields_is_datetime: HashMap, timestamp_field_opt: Option, + indexes_meta_for_leaf_search: &IndexesMetasForLeafSearch, ) -> crate::Result> { let index_uids = indexes_metadata .iter() @@ -1245,9 +1258,36 @@ async fn refine_and_list_matches( metastore, ) .await?; + + // Drop splits where every document falls before the index's retention cutoff. + let split_metadatas = split_metadatas + .into_iter() + .filter(|split| is_split_within_retention(split, indexes_meta_for_leaf_search)) + .collect(); + Ok(split_metadatas) } +/// Returns `false` only when a split's entire time range lies before the retention cutoff, +/// meaning every document it contains is expired. Splits without a time_range are always kept +/// since we cannot determine their age. +fn is_split_within_retention( + split: &SplitMetadata, + indexes_meta_for_leaf_search: &IndexesMetasForLeafSearch, +) -> bool { + let Some(index_meta) = indexes_meta_for_leaf_search.get(&split.index_uid) else { + return true; + }; + let Some(cutoff) = index_meta.retention_timestamp_cutoff_opt else { + return true; + }; + split + .time_range + .as_ref() + .map(|r| *r.end() >= cutoff) + .unwrap_or(true) +} + /// Fetches the list of splits and their metadata from the metastore async fn plan_splits_for_root_search( search_request: &mut SearchRequest, @@ -1270,7 +1310,12 @@ async fn plan_splits_for_root_search( return Ok((Vec::new(), HashMap::default())); } - let request_metadata = validate_request_and_build_metadata(&indexes_metadata, search_request)?; + let now_secs = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let request_metadata = + validate_request_and_build_metadata(&indexes_metadata, search_request, now_secs)?; let split_metadatas = refine_and_list_matches( metastore, search_request, @@ -1278,6 +1323,7 @@ async fn plan_splits_for_root_search( request_metadata.query_ast_resolved, request_metadata.sort_fields_is_datetime, request_metadata.timestamp_field_opt, + &request_metadata.indexes_meta_for_leaf_search, ) .await?; Ok(( @@ -1394,7 +1440,12 @@ pub async fn search_plan( ) .map_err(|err| SearchError::Internal(format!("failed to build doc mapper. cause: {err}")))?; - let request_metadata = validate_request_and_build_metadata(&indexes_metadata, &search_request)?; + let now_secs = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let request_metadata = + validate_request_and_build_metadata(&indexes_metadata, &search_request, now_secs)?; let split_metadatas = refine_and_list_matches( &mut metastore, &mut search_request, @@ -1402,6 +1453,7 @@ pub async fn search_plan( request_metadata.query_ast_resolved.clone(), request_metadata.sort_fields_is_datetime, request_metadata.timestamp_field_opt, + &request_metadata.indexes_meta_for_leaf_search, ) .await?; @@ -1835,6 +1887,7 @@ pub fn jobs_to_leaf_request( split_offsets: job_group.into_iter().map(|job| job.offsets).collect(), doc_mapper_ord, index_uri_ord, + retention_timestamp_cutoff: search_index_meta.retention_timestamp_cutoff_opt, }; leaf_search_request .leaf_requests @@ -2044,6 +2097,7 @@ mod tests { index_metadata_no_timestamp, ], &search_request, + 0, ) .unwrap(); assert_eq!( @@ -2099,6 +2153,7 @@ mod tests { let timestamp_field_different = validate_request_and_build_metadata( &[index_metadata_1, index_metadata_2], &search_request, + 0, ) .unwrap_err(); assert_eq!( @@ -2126,6 +2181,7 @@ mod tests { let timestamp_field_different = validate_request_and_build_metadata( &[index_metadata_1, index_metadata_2], &search_request, + 0, ) .unwrap_err(); assert_eq!( @@ -2204,6 +2260,7 @@ mod tests { let search_error = validate_request_and_build_metadata( &[index_metadata, index_metadata_with_other_config], &search_request, + 0, ) .unwrap_err(); assert_eq!( @@ -2212,6 +2269,79 @@ mod tests { ); } + #[test] + fn test_validate_request_and_build_metadata_retention_cutoff() { + let search_request = quickwit_proto::search::SearchRequest { + index_id_patterns: vec!["test-index".to_string()], + query_ast: qast_json_helper("test", &["body"]), + max_hits: 10, + ..Default::default() + }; + let mut index_metadata = IndexMetadata::for_test("test-index-1", "ram:///test-index-1"); + index_metadata.index_config.retention_policy_opt = Some(quickwit_config::RetentionPolicy { + retention_period: "7 days".to_string(), + evaluation_schedule: "hourly".to_string(), + }); + let now_secs: i64 = 7 * 24 * 3600 + 1000; + let request_metadata = + validate_request_and_build_metadata(&[index_metadata], &search_request, now_secs) + .unwrap(); + let index_uid = request_metadata + .indexes_meta_for_leaf_search + .keys() + .next() + .unwrap() + .clone(); + let cutoff = request_metadata + .indexes_meta_for_leaf_search + .get(&index_uid) + .unwrap() + .retention_timestamp_cutoff_opt; + assert_eq!(cutoff, Some(1000)); + } + + #[test] + fn test_split_pruning_by_retention_cutoff() { + let index_uid = quickwit_proto::types::IndexUid::for_test("test-index", 1); + let cutoff: i64 = 1_000; + + let index_meta = IndexMetasForLeafSearch { + index_uri: quickwit_common::uri::Uri::for_test("ram:///"), + doc_mapper_str: "{}".to_string(), + retention_timestamp_cutoff_opt: Some(cutoff), + }; + let indexes_meta: IndexesMetasForLeafSearch = + std::iter::once((index_uid.clone(), index_meta)).collect(); + + // entirely before cutoff + let mut split_before = + quickwit_metastore::SplitMetadata::for_test("before".to_string().into()); + split_before.index_uid = index_uid.clone(); + split_before.time_range = Some(0..=999); + assert!(!is_split_within_retention(&split_before, &indexes_meta)); + + // some docs before and after + let mut split_straddling = + quickwit_metastore::SplitMetadata::for_test("straddling".to_string().into()); + split_straddling.index_uid = index_uid.clone(); + split_straddling.time_range = Some(0..=1500); + assert!(is_split_within_retention(&split_straddling, &indexes_meta)); + + // all after + let mut split_after = + quickwit_metastore::SplitMetadata::for_test("after".to_string().into()); + split_after.index_uid = index_uid.clone(); + split_after.time_range = Some(1001..=2000); + assert!(is_split_within_retention(&split_after, &indexes_meta)); + + // no time range + let mut split_no_range = + quickwit_metastore::SplitMetadata::for_test("no_range".to_string().into()); + split_no_range.index_uid = index_uid.clone(); + split_no_range.time_range = None; + assert!(is_split_within_retention(&split_no_range, &indexes_meta)); + } + #[test] fn test_convert_sort_datetime_value() { let mut sort_value = SortValue::U64(1617000000000000000);